Refactor the SyncService to more closely implement the specification

.changeset/nervous-bobcats-grow.md

@@ -0,0 +1,5 @@
+---
+"@eth-optimism/l2geth": patch
+---
+
+Refactor the SyncService to more closely implement the specification. This includes using query params to select the backend from the DTL, trailing syncing of batches for the sequencer, syncing by batches as the verifier as well as unified code paths for transaction ingestion to prevent double ingestion or missed ingestion

l2geth/cmd/geth/main.go

@@ -165,6 +165,7 @@ var (
 		utils.RollupMaxCalldataSizeFlag,
 		utils.RollupDataPriceFlag,
 		utils.RollupExecutionPriceFlag,
+		utils.RollupBackendFlag,
 		utils.RollupEnableL2GasPollingFlag,
 		utils.RollupGasPriceOracleAddressFlag,
 		utils.RollupEnforceFeesFlag,

l2geth/cmd/geth/usage.go

@@ -80,6 +80,7 @@ var AppHelpFlagGroups = []flagGroup{
 			utils.RollupMaxCalldataSizeFlag,
 			utils.RollupDataPriceFlag,
 			utils.RollupExecutionPriceFlag,
+			utils.RollupBackendFlag,
 			utils.RollupEnableL2GasPollingFlag,
 			utils.RollupGasPriceOracleAddressFlag,
 			utils.RollupEnforceFeesFlag,

l2geth/cmd/utils/flags.go

@@ -851,6 +851,12 @@ var (
 		Value:  time.Minute * 3,
 		EnvVar: "ROLLUP_TIMESTAMP_REFRESH",
 	}
+	RollupBackendFlag = cli.StringFlag{
+		Name:   "rollup.backend",
+		Usage:  "Sync backend for verifiers (\"l1\" or \"l2\"), defaults to l1",
+		Value:  "l1",
+		EnvVar: "ROLLUP_BACKEND",
+	}
 	// Flag to enable verifier mode
 	RollupEnableVerifierFlag = cli.BoolFlag{
 		Name:   "rollup.verifier",
@@ -1188,6 +1194,15 @@ func setRollup(ctx *cli.Context, cfg *rollup.Config) {
 	if ctx.GlobalIsSet(RollupExecutionPriceFlag.Name) {
 		cfg.ExecutionPrice = GlobalBig(ctx, RollupExecutionPriceFlag.Name)
 	}
+	if ctx.GlobalIsSet(RollupBackendFlag.Name) {
+		val := ctx.GlobalString(RollupBackendFlag.Name)
+		backend, err := rollup.NewBackend(val)
+		if err != nil {
+			log.Error("Configured with unknown sync backend, defaulting to l1", "backend", val)
+			backend, _ = rollup.NewBackend("l1")
+		}
+		cfg.Backend = backend
+	}
 	if ctx.GlobalIsSet(RollupGasPriceOracleAddressFlag.Name) {
 		addr := ctx.GlobalString(RollupGasPriceOracleAddressFlag.Name)
 		cfg.GasPriceOracleAddress = common.HexToAddress(addr)

l2geth/core/rawdb/rollup_indexes.go

@@ -69,3 +69,24 @@ func WriteHeadVerifiedIndex(db ethdb.KeyValueWriter, index uint64) {
 		log.Crit("Failed to store verifier index", "err", err)
 	}
 }
+
+// ReadHeadBatchIndex will read the known tip of the processed batches
+func ReadHeadBatchIndex(db ethdb.KeyValueReader) *uint64 {
+	data, _ := db.Get(headBatchKey)
+	if len(data) == 0 {
+		return nil
+	}
+	ret := new(big.Int).SetBytes(data).Uint64()
+	return &ret
+}
+
+// WriteHeadBatchIndex will write the known tip of the processed batches
+func WriteHeadBatchIndex(db ethdb.KeyValueWriter, index uint64) {
+	value := new(big.Int).SetUint64(index).Bytes()
+	if index == 0 {
+		value = []byte{0}
+	}
+	if err := db.Put(headBatchKey, value); err != nil {
+		log.Crit("Failed to store head batch index", "err", err)
+	}
+}

l2geth/core/rawdb/schema.go

@@ -62,6 +62,8 @@ var (
 	headQueueIndexKey = []byte("LastQueueIndex")
 	// headVerifiedIndexKey tracks the latest verified index
 	headVerifiedIndexKey = []byte("LastVerifiedIndex")
+	// headBatchKey tracks the latest processed batch
+	headBatchKey = []byte("LastBatch")
 
 	preimagePrefix = []byte("secure-key-")      // preimagePrefix + hash -> preimage
 	configPrefix   = []byte("ethereum-config-") // config prefix for the db

l2geth/eth/api_backend.go

@@ -330,7 +330,7 @@ func (b *EthAPIBackend) SendTx(ctx context.Context, signedTx *types.Transaction)
 				}
 			}
 		}
-		return b.eth.syncService.ApplyTransaction(signedTx)
+		return b.eth.syncService.ValidateAndApplySequencerTransaction(signedTx)
 	}
 	// OVM Disabled
 	return b.eth.txPool.AddLocal(signedTx)

l2geth/rollup/client.go

@@ -116,14 +116,17 @@ type decoded struct {
 type RollupClient interface {
 	GetEnqueue(index uint64) (*types.Transaction, error)
 	GetLatestEnqueue() (*types.Transaction, error)
-	GetTransaction(uint64) (*types.Transaction, error)
-	GetLatestTransaction() (*types.Transaction, error)
+	GetLatestEnqueueIndex() (*uint64, error)
+	GetTransaction(uint64, Backend) (*types.Transaction, error)
+	GetLatestTransaction(Backend) (*types.Transaction, error)
+	GetLatestTransactionIndex(Backend) (*uint64, error)
 	GetEthContext(uint64) (*EthContext, error)
 	GetLatestEthContext() (*EthContext, error)
 	GetLastConfirmedEnqueue() (*types.Transaction, error)
 	GetLatestTransactionBatch() (*Batch, []*types.Transaction, error)
+	GetLatestTransactionBatchIndex() (*uint64, error)
 	GetTransactionBatch(uint64) (*Batch, []*types.Transaction, error)
-	SyncStatus() (*SyncStatus, error)
+	SyncStatus(Backend) (*SyncStatus, error)
 	GetL1GasPrice() (*big.Int, error)
 }
 
@@ -270,6 +273,43 @@ func (c *Client) GetLatestEnqueue() (*types.Transaction, error) {
 	return tx, nil
 }
 
+// GetLatestEnqueueIndex returns the latest `enqueue()` index
+func (c *Client) GetLatestEnqueueIndex() (*uint64, error) {
+	tx, err := c.GetLatestEnqueue()
+	if err != nil {
+		return nil, err
+	}
+	index := tx.GetMeta().QueueIndex
+	if index == nil {
+		return nil, errors.New("Latest queue index is nil")
+	}
+	return index, nil
+}
+
+// GetLatestTransactionIndex returns the latest CTC index that has been batch
+// submitted or not, depending on the backend
+func (c *Client) GetLatestTransactionIndex(backend Backend) (*uint64, error) {
+	tx, err := c.GetLatestTransaction(backend)
+	if err != nil {
+		return nil, err
+	}
+	index := tx.GetMeta().Index
+	if index == nil {
+		return nil, errors.New("Latest index is nil")
+	}
+	return index, nil
+}
+
+// GetLatestTransactionBatchIndex returns the latest transaction batch index
+func (c *Client) GetLatestTransactionBatchIndex() (*uint64, error) {
+	batch, _, err := c.GetLatestTransactionBatch()
+	if err != nil {
+		return nil, err
+	}
+	index := batch.Index
+	return &index, nil
+}
+
 // batchedTransactionToTransaction converts a transaction into a
 // types.Transaction that can be consumed by the SyncService
 func batchedTransactionToTransaction(res *transaction, signer *types.EIP155Signer) (*types.Transaction, error) {
@@ -364,12 +404,15 @@ func batchedTransactionToTransaction(res *transaction, signer *types.EIP155Signe
 }
 
 // GetTransaction will get a transaction by Canonical Transaction Chain index
-func (c *Client) GetTransaction(index uint64) (*types.Transaction, error) {
+func (c *Client) GetTransaction(index uint64, backend Backend) (*types.Transaction, error) {
 	str := strconv.FormatUint(index, 10)
 	response, err := c.client.R().
 		SetPathParams(map[string]string{
 			"index": str,
 		}).
+		SetQueryParams(map[string]string{
+			"backend": backend.String(),
+		}).
 		SetResult(&TransactionResponse{}).
 		Get("/transaction/index/{index}")
 
@@ -385,9 +428,12 @@ func (c *Client) GetTransaction(index uint64) (*types.Transaction, error) {
 
 // GetLatestTransaction will get the latest transaction, meaning the transaction
 // with the greatest Canonical Transaction Chain index
-func (c *Client) GetLatestTransaction() (*types.Transaction, error) {
+func (c *Client) GetLatestTransaction(backend Backend) (*types.Transaction, error) {
 	response, err := c.client.R().
 		SetResult(&TransactionResponse{}).
+		SetQueryParams(map[string]string{
+			"backend": backend.String(),
+		}).
 		Get("/transaction/latest")
 
 	if err != nil {
@@ -477,9 +523,12 @@ func (c *Client) GetLastConfirmedEnqueue() (*types.Transaction, error) {
 }
 
 // SyncStatus will query the remote server to determine if it is still syncing
-func (c *Client) SyncStatus() (*SyncStatus, error) {
+func (c *Client) SyncStatus(backend Backend) (*SyncStatus, error) {
 	response, err := c.client.R().
 		SetResult(&SyncStatus{}).
+		SetQueryParams(map[string]string{
+			"backend": backend.String(),
+		}).
 		Get("/eth/syncing")
 
 	if err != nil {
@@ -533,8 +582,8 @@ func (c *Client) GetTransactionBatch(index uint64) (*Batch, []*types.Transaction
 // parseTransactionBatchResponse will turn a TransactionBatchResponse into a
 // Batch and its corresponding types.Transactions
 func parseTransactionBatchResponse(txBatch *TransactionBatchResponse, signer *types.EIP155Signer) (*Batch, []*types.Transaction, error) {
-	if txBatch == nil {
-		return nil, nil, nil
+	if txBatch == nil || txBatch.Batch == nil {
+		return nil, nil, errElementNotFound
 	}
 	batch := txBatch.Batch
 	txs := make([]*types.Transaction, len(txBatch.Transactions))

l2geth/rollup/config.go

@@ -40,6 +40,8 @@ type Config struct {
 	DataPrice *big.Int
 	// The gas price to use for L2 congestion costs
 	ExecutionPrice *big.Int
+	// Represents the source of the transactions that is being synced
+	Backend Backend
 	// Only accept transactions with fees
 	EnforceFees bool
 }