indexer: Test preparation (#3657)

Prepares the repo to land the indexer changes by mocing some additional utilities to `op-e2e` and moving `backoff` from `op-node` into `op-service`.

op-e2e/e2eutils/testing.go

 package e2eutils
 
+import (
+	"context"
+	"testing"
+	"time"
+)
+
 // TestingBase is an interface used for standard Go testing.
 // This interface is used for unit tests, benchmarks, and fuzz tests and also emulated in Hive.
 //
@@ -24,3 +30,9 @@ type TestingBase interface {
 	Skipped() bool
 	TempDir() string
 }
+
+func TimeoutCtx(t *testing.T, timeout time.Duration) context.Context {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	return ctx
+}

op-e2e/e2eutils/waits.go

+package e2eutils
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethclient"
+)
+
+func WaitReceiptOK(ctx context.Context, client *ethclient.Client, hash common.Hash) (*types.Receipt, error) {
+	return WaitReceipt(ctx, client, hash, types.ReceiptStatusSuccessful)
+}
+
+func WaitReceiptFail(ctx context.Context, client *ethclient.Client, hash common.Hash) (*types.Receipt, error) {
+	return WaitReceipt(ctx, client, hash, types.ReceiptStatusFailed)
+}
+
+func WaitReceipt(ctx context.Context, client *ethclient.Client, hash common.Hash, status uint64) (*types.Receipt, error) {
+	ticker := time.NewTicker(100 * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		receipt, err := client.TransactionReceipt(ctx, hash)
+		if errors.Is(err, ethereum.NotFound) {
+			select {
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			case <-ticker.C:
+				continue
+			}
+		}
+		if err != nil {
+			return nil, err
+		}
+		if receipt.Status != status {
+			return receipt, fmt.Errorf("expected status %d, but got %d", status, receipt.Status)
+		}
+		return receipt, nil
+	}
+}
+
+func WaitBlock(ctx context.Context, client *ethclient.Client, n uint64) error {
+	for {
+		height, err := client.BlockNumber(ctx)
+		if err != nil {
+			return err
+		}
+		if height < n {
+			time.Sleep(500 * time.Millisecond)
+			continue
+		}
+		break
+	}
+
+	return nil
+}
+
+func WaitFor(ctx context.Context, rate time.Duration, cb func() (bool, error)) error {
+	tick := time.NewTicker(rate)
+	defer tick.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-tick.C:
+			done, err := cb()
+			if err != nil {
+				return err
+			}
+			if done {
+				return nil
+			}
+		}
+	}
+}

op-e2e/geth.go

@@ -155,6 +155,8 @@ func initL2Geth(name string, l2ChainID *big.Int, genesis *core.Genesis, jwtPath
 		WSPort:      0,
 		AuthAddr:    "127.0.0.1",
 		AuthPort:    0,
+		HTTPHost:    "127.0.0.1",
+		HTTPPort:    0,
 		WSModules:   []string{"debug", "admin", "eth", "txpool", "net", "rpc", "web3", "personal", "engine"},
 		HTTPModules: []string{"debug", "admin", "eth", "txpool", "net", "rpc", "web3", "personal", "engine"},
 		JWTSecret:   jwtPath,

op-e2e/setup.go

@@ -190,27 +190,27 @@ type System struct {
 	RollupConfig *rollup.Config
 
 	// Connections to running nodes
-	nodes             map[string]*node.Node
-	backends          map[string]*geth_eth.Ethereum
+	Nodes             map[string]*node.Node
+	Backends          map[string]*geth_eth.Ethereum
 	Clients           map[string]*ethclient.Client
-	rollupNodes       map[string]*rollupNode.OpNode
-	l2OutputSubmitter *l2os.L2OutputSubmitter
-	batchSubmitter    *bss.BatchSubmitter
+	RollupNodes       map[string]*rollupNode.OpNode
+	L2OutputSubmitter *l2os.L2OutputSubmitter
+	BatchSubmitter    *bss.BatchSubmitter
 	Mocknet           mocknet.Mocknet
 }
 
 func (sys *System) Close() {
-	if sys.l2OutputSubmitter != nil {
-		sys.l2OutputSubmitter.Stop()
+	if sys.L2OutputSubmitter != nil {
+		sys.L2OutputSubmitter.Stop()
 	}
-	if sys.batchSubmitter != nil {
-		sys.batchSubmitter.Stop()
+	if sys.BatchSubmitter != nil {
+		sys.BatchSubmitter.Stop()
 	}
 
-	for _, node := range sys.rollupNodes {
+	for _, node := range sys.RollupNodes {
 		node.Close()
 	}
-	for _, node := range sys.nodes {
+	for _, node := range sys.Nodes {
 		node.Close()
 	}
 	sys.Mocknet.Close()
@@ -219,18 +219,18 @@ func (sys *System) Close() {
 func (cfg SystemConfig) Start() (*System, error) {
 	sys := &System{
 		cfg:         cfg,
-		nodes:       make(map[string]*node.Node),
-		backends:    make(map[string]*geth_eth.Ethereum),
+		Nodes:       make(map[string]*node.Node),
+		Backends:    make(map[string]*geth_eth.Ethereum),
 		Clients:     make(map[string]*ethclient.Client),
-		rollupNodes: make(map[string]*rollupNode.OpNode),
+		RollupNodes: make(map[string]*rollupNode.OpNode),
 	}
 	didErrAfterStart := false
 	defer func() {
 		if didErrAfterStart {
-			for _, node := range sys.rollupNodes {
+			for _, node := range sys.RollupNodes {
 				node.Close()
 			}
-			for _, node := range sys.nodes {
+			for _, node := range sys.Nodes {
 				node.Close()
 			}
 		}
@@ -302,16 +302,16 @@ func (cfg SystemConfig) Start() (*System, error) {
 	if err != nil {
 		return nil, err
 	}
-	sys.nodes["l1"] = l1Node
-	sys.backends["l1"] = l1Backend
+	sys.Nodes["l1"] = l1Node
+	sys.Backends["l1"] = l1Backend
 
 	for name := range cfg.Nodes {
 		node, backend, err := initL2Geth(name, big.NewInt(int64(cfg.DeployConfig.L2ChainID)), l2Genesis, cfg.JWTFilePath)
 		if err != nil {
 			return nil, err
 		}
-		sys.nodes[name] = node
-		sys.backends[name] = backend
+		sys.Nodes[name] = node
+		sys.Backends[name] = backend
 	}
 
 	// Start
@@ -325,7 +325,7 @@ func (cfg SystemConfig) Start() (*System, error) {
 		didErrAfterStart = true
 		return nil, err
 	}
-	for name, node := range sys.nodes {
+	for name, node := range sys.Nodes {
 		if name == "l1" {
 			continue
 		}
@@ -347,9 +347,9 @@ func (cfg SystemConfig) Start() (*System, error) {
 	}
 
 	for name, rollupCfg := range cfg.Nodes {
-		l2EndpointConfig := sys.nodes[name].WSAuthEndpoint()
+		l2EndpointConfig := sys.Nodes[name].WSAuthEndpoint()
 		if useHTTP {
-			l2EndpointConfig = sys.nodes[name].HTTPAuthEndpoint()
+			l2EndpointConfig = sys.Nodes[name].HTTPAuthEndpoint()
 		}
 		rollupCfg.L1 = &rollupNode.L1EndpointConfig{
 			L1NodeAddr: l1EndpointConfig,
@@ -371,7 +371,7 @@ func (cfg SystemConfig) Start() (*System, error) {
 	}
 	l1Client := ethclient.NewClient(rpc.DialInProc(l1Srv))
 	sys.Clients["l1"] = l1Client
-	for name, node := range sys.nodes {
+	for name, node := range sys.Nodes {
 		client, err := ethclient.DialContext(ctx, node.WSEndpoint())
 		if err != nil {
 			didErrAfterStart = true
@@ -459,7 +459,7 @@ func (cfg SystemConfig) Start() (*System, error) {
 			didErrAfterStart = true
 			return nil, err
 		}
-		sys.rollupNodes[name] = node
+		sys.RollupNodes[name] = node
 	}
 
 	if cfg.P2PTopology != nil {
@@ -484,10 +484,10 @@ func (cfg SystemConfig) Start() (*System, error) {
 	}
 
 	// L2Output Submitter
-	sys.l2OutputSubmitter, err = l2os.NewL2OutputSubmitter(l2os.Config{
-		L1EthRpc:                  sys.nodes["l1"].WSEndpoint(),
-		L2EthRpc:                  sys.nodes["sequencer"].WSEndpoint(),
-		RollupRpc:                 sys.rollupNodes["sequencer"].HTTPEndpoint(),
+	sys.L2OutputSubmitter, err = l2os.NewL2OutputSubmitter(l2os.Config{
+		L1EthRpc:                  sys.Nodes["l1"].WSEndpoint(),
+		L2EthRpc:                  sys.Nodes["sequencer"].WSEndpoint(),
+		RollupRpc:                 sys.RollupNodes["sequencer"].HTTPEndpoint(),
 		L2OOAddress:               predeploys.DevL2OutputOracleAddr.String(),
 		PollInterval:              50 * time.Millisecond,
 		NumConfirmations:          1,
@@ -503,15 +503,15 @@ func (cfg SystemConfig) Start() (*System, error) {
 		return nil, fmt.Errorf("unable to setup l2 output submitter: %w", err)
 	}
 
-	if err := sys.l2OutputSubmitter.Start(); err != nil {
+	if err := sys.L2OutputSubmitter.Start(); err != nil {
 		return nil, fmt.Errorf("unable to start l2 output submitter: %w", err)
 	}
 
 	// Batch Submitter
-	sys.batchSubmitter, err = bss.NewBatchSubmitter(bss.Config{
-		L1EthRpc:                  sys.nodes["l1"].WSEndpoint(),
-		L2EthRpc:                  sys.nodes["sequencer"].WSEndpoint(),
-		RollupRpc:                 sys.rollupNodes["sequencer"].HTTPEndpoint(),
+	sys.BatchSubmitter, err = bss.NewBatchSubmitter(bss.Config{
+		L1EthRpc:                  sys.Nodes["l1"].WSEndpoint(),
+		L2EthRpc:                  sys.Nodes["sequencer"].WSEndpoint(),
+		RollupRpc:                 sys.RollupNodes["sequencer"].HTTPEndpoint(),
 		MinL1TxSize:               1,
 		MaxL1TxSize:               120000,
 		ChannelTimeout:            cfg.DeployConfig.ChannelTimeout,
@@ -530,7 +530,7 @@ func (cfg SystemConfig) Start() (*System, error) {
 		return nil, fmt.Errorf("failed to setup batch submitter: %w", err)
 	}
 
-	if err := sys.batchSubmitter.Start(); err != nil {
+	if err := sys.BatchSubmitter.Start(); err != nil {
 		return nil, fmt.Errorf("unable to start batch submitter: %w", err)
 	}
 

op-e2e/system_test.go

@@ -55,7 +55,7 @@ func TestL2OutputSubmitter(t *testing.T) {
 
 	l1Client := sys.Clients["l1"]
 
-	rollupRPCClient, err := rpc.DialContext(context.Background(), sys.rollupNodes["sequencer"].HTTPEndpoint())
+	rollupRPCClient, err := rpc.DialContext(context.Background(), sys.RollupNodes["sequencer"].HTTPEndpoint())
 	require.Nil(t, err)
 	rollupClient := sources.NewRollupClient(rollupRPCClient)
 
@@ -210,7 +210,7 @@ func TestSystemE2E(t *testing.T) {
 	require.Equal(t, verifBlock.ParentHash(), seqBlock.ParentHash(), "Verifier and sequencer blocks parent hashes not the same after including a batch tx")
 	require.Equal(t, verifBlock.Hash(), seqBlock.Hash(), "Verifier and sequencer blocks not the same after including a batch tx")
 
-	rollupRPCClient, err := rpc.DialContext(context.Background(), sys.rollupNodes["sequencer"].HTTPEndpoint())
+	rollupRPCClient, err := rpc.DialContext(context.Background(), sys.RollupNodes["sequencer"].HTTPEndpoint())
 	require.Nil(t, err)
 	rollupClient := sources.NewRollupClient(rollupRPCClient)
 	// basic check that sync status works
@@ -317,7 +317,7 @@ func TestMintOnRevertedDeposit(t *testing.T) {
 	// Find deposit contract
 	depositContract, err := bindings.NewOptimismPortal(predeploys.DevOptimismPortalAddr, l1Client)
 	require.Nil(t, err)
-	l1Node := sys.nodes["l1"]
+	l1Node := sys.Nodes["l1"]
 
 	// create signer
 	ks := l1Node.AccountManager().Backends(keystore.KeyStoreType)[0].(*keystore.KeyStore)
@@ -790,7 +790,7 @@ func TestWithdrawals(t *testing.T) {
 	header, err = l2Verif.HeaderByNumber(ctx, new(big.Int).SetUint64(blockNumber))
 	require.Nil(t, err)
 
-	rpc, err := rpc.Dial(sys.nodes["verifier"].WSEndpoint())
+	rpc, err := rpc.Dial(sys.Nodes["verifier"].WSEndpoint())
 	require.Nil(t, err)
 	l2client := withdrawals.NewClient(rpc)
 

op-node/client/rpc.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"regexp"
 
-	"github.com/ethereum-optimism/optimism/op-node/backoff"
+	"github.com/ethereum-optimism/optimism/op-service/backoff"
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/prometheus/client_golang/prometheus"

op-node/rollup/driver/state.go

@@ -11,10 +11,10 @@ import (
 
 	"github.com/ethereum/go-ethereum/log"
 
-	"github.com/ethereum-optimism/optimism/op-node/backoff"
 	"github.com/ethereum-optimism/optimism/op-node/eth"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+	"github.com/ethereum-optimism/optimism/op-service/backoff"
 )
 
 // Deprecated: use eth.SyncStatus instead.

op-node/backoff/operation.go → op-service/backoff/operation.go

 package backoff
 
 import (
+	"context"
 	"fmt"
 	"time"
 )
@@ -24,21 +25,37 @@ func (e *ErrFailedPermanently) Error() string {
 // with delays in between each retry according to the provided
 // Strategy.
 func Do(maxAttempts int, strategy Strategy, op Operation) error {
+	return DoCtx(context.Background(), maxAttempts, strategy, op)
+}
+
+func DoCtx(ctx context.Context, maxAttempts int, strategy Strategy, op Operation) error {
 	var attempt int
 
+	reattemptCh := make(chan struct{}, 1)
+	doReattempt := func() {
+		reattemptCh <- struct{}{}
+	}
+	doReattempt()
+
 	for {
-		attempt++
-		err := op()
-		if err == nil {
-			return nil
-		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-reattemptCh:
+			attempt++
+			err := op()
+			if err == nil {
+				return nil
+			}
 
-		if attempt == maxAttempts {
-			return &ErrFailedPermanently{
-				attempts: maxAttempts,
-				LastErr:  err,
+			if attempt == maxAttempts {
+				return &ErrFailedPermanently{
+					attempts: maxAttempts,
+					LastErr:  err,
+				}
 			}
+			time.AfterFunc(strategy.Duration(attempt-1), doReattempt)
 		}
-		time.Sleep(strategy.Duration(attempt - 1))
+
 	}
 }