Merge branch 'develop' of https://github.com/epociask/optimism into indexer.api.supply_view

.foundryrc

-3b1129b5bc43ba22a9bcf4e4323c5a9df0023140
+d85718785859dc0b5a095d2302d1a20ec06ab77a

.gethrc

+v1.13.4

.github/CODEOWNERS

@@ -16,7 +16,7 @@
 /op-exporter    @ethereum-optimism/go-reviewers
 /op-heartbeat   @ethereum-optimism/go-reviewers
 /op-node        @ethereum-optimism/go-reviewers
-/op-node/rollup @protolambda @trianglesphere
+/op-node/rollup @protolambda @trianglesphere @ajsutton
 /op-preimage    @ethereum-optimism/go-reviewers
 /op-program     @ethereum-optimism/go-reviewers
 /op-proposer    @ethereum-optimism/go-reviewers
@@ -28,6 +28,7 @@
 /.circleci       @ethereum-optimism/infra-reviewers
 /.github         @ethereum-optimism/infra-reviewers
 /ops             @ethereum-optimism/infra-reviewers
+/docker-bake.hcl @ethereum-optimism/infra-reviewers
 
 # Misc
 /proxyd           @ethereum-optimism/infra-reviewers

.gitmodules

@@ -17,6 +17,3 @@
 	path = packages/contracts-bedrock/lib/safe-contracts
 	url = https://github.com/safe-global/safe-contracts
 	branch = v1.4.0
-
-
-

.nvmrc

-v18.17.1
+v20.9.0

CONTRIBUTING.md

@@ -181,9 +181,9 @@ You must have Python 3.x installed to run `slither`.
 To run `slither` locally, do:
 
 ```bash
-cd packages/contracts
+cd packages/contracts-bedrock
 pip3 install slither-analyzer
-pnpm test:slither
+pnpm slither
 ```
 
 ## Labels

Makefile

 COMPOSEFLAGS=-d
 ITESTS_L2_HOST=http://localhost:9545
 BEDROCK_TAGS_REMOTE?=origin
+OP_STACK_GO_BUILDER?=us-docker.pkg.dev/oplabs-tools-artifacts/images/op-stack-go:latest
+
+# Requires at least Python v3.9; specify a minor version below if needed
+PYTHON?=python3
 
 build: build-go build-ts
 .PHONY: build
@@ -8,6 +12,10 @@ build: build-go build-ts
 build-go: submodules op-node op-proposer op-batcher
 .PHONY: build-go
 
+lint-go:
+	golangci-lint run -E goimports,sqlclosecheck,bodyclose,asciicheck,misspell,errorlint --timeout 5m -e "errors.As" -e "errors.Is" ./...
+.PHONY: lint-go
+
 build-ts: submodules
 	if [ -n "$$NVM_DIR" ]; then \
 		. $$NVM_DIR/nvm.sh && nvm use; \
@@ -19,6 +27,18 @@ build-ts: submodules
 ci-builder:
 	docker build -t ci-builder -f ops/docker/ci-builder/Dockerfile .
 
+golang-docker:
+	# We don't use a buildx builder here, and just load directly into regular docker, for convenience.
+	GIT_COMMIT=$$(git rev-parse HEAD) \
+	GIT_DATE=$$(git show -s --format='%ct') \
+	IMAGE_TAGS=$$(git rev-parse HEAD),latest \
+	docker buildx bake \
+			--progress plain \
+			--load \
+			-f docker-bake.hcl \
+			op-node op-batcher op-proposer op-challenger
+.PHONY: golang-docker
+
 submodules:
 	# CI will checkout submodules on its own (and fails on these commands)
 	if [ -z "$$GITHUB_ENV" ]; then \
@@ -97,14 +117,14 @@ pre-devnet:
 devnet-up: pre-devnet
 	./ops/scripts/newer-file.sh .devnet/allocs-l1.json ./packages/contracts-bedrock \
 		|| make devnet-allocs
-	PYTHONPATH=./bedrock-devnet python3 ./bedrock-devnet/main.py --monorepo-dir=.
+	PYTHONPATH=./bedrock-devnet $(PYTHON) ./bedrock-devnet/main.py --monorepo-dir=.
 .PHONY: devnet-up
 
 # alias for devnet-up
 devnet-up-deploy: devnet-up
 
 devnet-test: pre-devnet
-	PYTHONPATH=./bedrock-devnet python3 ./bedrock-devnet/main.py --monorepo-dir=. --test
+	PYTHONPATH=./bedrock-devnet $(PYTHON) ./bedrock-devnet/main.py --monorepo-dir=. --test
 .PHONY: devnet-test
 
 devnet-down:
@@ -120,7 +140,7 @@ devnet-clean:
 .PHONY: devnet-clean
 
 devnet-allocs: pre-devnet
-	PYTHONPATH=./bedrock-devnet python3 ./bedrock-devnet/main.py --monorepo-dir=. --allocs
+	PYTHONPATH=./bedrock-devnet $(PYTHON) ./bedrock-devnet/main.py --monorepo-dir=. --allocs
 
 devnet-logs:
 	@(cd ./ops-bedrock && docker compose logs -f)
@@ -163,4 +183,9 @@ bedrock-markdown-links:
 		--exclude-mail /input/README.md "/input/specs/**/*.md"
 
 install-geth:
-	go install github.com/ethereum/go-ethereum/cmd/geth@v1.12.0
+	./ops/scripts/geth-version-checker.sh && \
+	 	(echo "Geth versions match, not installing geth..."; true) || \
+ 		(echo "Versions do not match, installing geth!"; \
+ 			go install -v github.com/ethereum/go-ethereum/cmd/geth@$(shell cat .gethrc); \
+ 			echo "Installed geth!"; true)
+.PHONY: install-geth

bedrock-devnet/devnet/__init__.py

@@ -10,6 +10,9 @@ import time
 import shutil
 import http.client
 from multiprocessing import Process, Queue
+import concurrent.futures
+from collections import namedtuple
+
 
 import devnet.log_setup
 
@@ -94,9 +97,20 @@ def main():
         devnet_l1_genesis(paths)
         return
 
-    log.info('Building docker images')
-    run_command(['docker', 'compose', 'build', '--progress', 'plain'], cwd=paths.ops_bedrock_dir, env={
-        'PWD': paths.ops_bedrock_dir
+    git_commit = subprocess.run(['git', 'rev-parse', 'HEAD'], capture_output=True, text=True).stdout.strip()
+    git_date = subprocess.run(['git', 'show', '-s', "--format=%ct"], capture_output=True, text=True).stdout.strip()
+
+    # CI loads the images from workspace, and does not otherwise know the images are good as-is
+    if os.getenv('DEVNET_NO_BUILD') == "true":
+        log.info('Skipping docker images build')
+    else:
+        log.info(f'Building docker images for git commit {git_commit} ({git_date})')
+        run_command(['docker', 'compose', 'build', '--progress', 'plain',
+                     '--build-arg', f'GIT_COMMIT={git_commit}', '--build-arg', f'GIT_DATE={git_date}'],
+                    cwd=paths.ops_bedrock_dir, env={
+            'PWD': paths.ops_bedrock_dir,
+            'DOCKER_BUILDKIT': '1', # (should be available by default in later versions, but explicitly enable it anyway)
+            'COMPOSE_DOCKER_CLI_BUILD': '1'  # use the docker cache
         })
 
     log.info('Devnet starting')
@@ -188,13 +202,12 @@ def devnet_deploy(paths):
         # If someone reads this comment and understands why this is being done, please
         # update this comment to explain.
         init_devnet_l1_deploy_config(paths, update_timestamp=True)
-        outfile_l1 = pjoin(paths.devnet_dir, 'genesis-l1.json')
         run_command([
             'go', 'run', 'cmd/main.go', 'genesis', 'l1',
             '--deploy-config', paths.devnet_config_path,
             '--l1-allocs', paths.allocs_path,
             '--l1-deployments', paths.addresses_json_path,
-            '--outfile.l1', outfile_l1,
+            '--outfile.l1', paths.genesis_l1_path,
         ], cwd=paths.op_node_dir)
 
     log.info('Starting L1.')
@@ -213,8 +226,8 @@ def devnet_deploy(paths):
             '--l1-rpc', 'http://localhost:8545',
             '--deploy-config', paths.devnet_config_path,
             '--deployment-dir', paths.deployment_dir,
-            '--outfile.l2', pjoin(paths.devnet_dir, 'genesis-l2.json'),
-            '--outfile.rollup', pjoin(paths.devnet_dir, 'rollup.json')
+            '--outfile.l2', paths.genesis_l2_path,
+            '--outfile.rollup', paths.rollup_config_path
         ], cwd=paths.op_node_dir)
 
     rollup_config = read_json(paths.rollup_config_path)
@@ -274,21 +287,27 @@ def debug_dumpBlock(url):
 def wait_for_rpc_server(url):
     log.info(f'Waiting for RPC server at {url}')
 
-    conn = http.client.HTTPConnection(url)
     headers = {'Content-type': 'application/json'}
     body = '{"id":1, "jsonrpc":"2.0", "method": "eth_chainId", "params":[]}'
 
     while True:
         try:
+            conn = http.client.HTTPConnection(url)
             conn.request('POST', '/', body, headers)
             response = conn.getresponse()
-            conn.close()
             if response.status < 300:
                 log.info(f'RPC server at {url} ready')
                 return
         except Exception as e:
             log.info(f'Waiting for RPC server at {url}')
             time.sleep(1)
+        finally:
+            if conn:
+                conn.close()
+
+
+CommandPreset = namedtuple('Command', ['name', 'args', 'cwd', 'timeout'])
+
 
 def devnet_test(paths):
     # Check the L2 config
@@ -297,17 +316,57 @@ def devnet_test(paths):
         cwd=paths.ops_chain_ops,
     )
 
-    run_command(
-         ['npx', 'hardhat',  'deposit-erc20', '--network',  'devnetL1', '--l1-contracts-json-path', paths.addresses_json_path],
-         cwd=paths.sdk_dir,
-         timeout=8*60,
-    )
+    # Run the two commands with different signers, so the ethereum nonce management does not conflict
+    # And do not use devnet system addresses, to avoid breaking fee-estimation or nonce values.
+    run_commands([
+        CommandPreset('erc20-test',
+          ['npx', 'hardhat',  'deposit-erc20', '--network',  'devnetL1',
+           '--l1-contracts-json-path', paths.addresses_json_path, '--signer-index', '14'],
+          cwd=paths.sdk_dir, timeout=8*60),
+        CommandPreset('eth-test',
+          ['npx', 'hardhat',  'deposit-eth', '--network',  'devnetL1',
+           '--l1-contracts-json-path', paths.addresses_json_path, '--signer-index', '15'],
+          cwd=paths.sdk_dir, timeout=8*60)
+    ], max_workers=2)
+
+
+def run_commands(commands: list[CommandPreset], max_workers=2):
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+        futures = [executor.submit(run_command_preset, cmd) for cmd in commands]
+
+        for future in concurrent.futures.as_completed(futures):
+            result = future.result()
+            if result:
+                print(result.stdout)
+
+
+def run_command_preset(command: CommandPreset):
+    with subprocess.Popen(command.args, cwd=command.cwd,
+                          stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) as proc:
+        try:
+            # Live output processing
+            for line in proc.stdout:
+                # Annotate and print the line with timestamp and command name
+                timestamp = datetime.datetime.utcnow().strftime('%H:%M:%S.%f')
+                # Annotate and print the line with the timestamp
+                print(f"[{timestamp}][{command.name}] {line}", end='')
+
+            stdout, stderr = proc.communicate(timeout=command.timeout)
+
+            if proc.returncode != 0:
+                raise RuntimeError(f"Command '{' '.join(command.args)}' failed with return code {proc.returncode}: {stderr}")
+
+        except subprocess.TimeoutExpired:
+            raise RuntimeError(f"Command '{' '.join(command.args)}' timed out!")
+
+        except Exception as e:
+            raise RuntimeError(f"Error executing '{' '.join(command.args)}': {e}")
+
+        finally:
+            # Ensure process is terminated
+            proc.kill()
+    return proc.returncode
 
-    run_command(
-         ['npx', 'hardhat',  'deposit-eth', '--network',  'devnetL1', '--l1-contracts-json-path', paths.addresses_json_path],
-         cwd=paths.sdk_dir,
-         timeout=8*60,
-    )
 
 def run_command(args, check=True, shell=False, cwd=None, env=None, timeout=None):
     env = env if env else {}

cannon/Makefile

-GITCOMMIT := $(shell git rev-parse HEAD)
-GITDATE := $(shell git show -s --format='%ct')
+GITCOMMIT ?= $(shell git rev-parse HEAD)
+GITDATE ?= $(shell git show -s --format='%ct')
 VERSION := v0.0.0
 
 LDFLAGSSTRING +=-X main.GitCommit=$(GITCOMMIT)
@@ -20,9 +20,6 @@ elf:
 test: elf
 	go test -v ./...
 
-lint:
-	golangci-lint run -E goimports,sqlclosecheck,bodyclose,asciicheck,misspell,errorlint --timeout 5m -e "errors.As" -e "errors.Is"
-
 fuzz:
 	go test -run NOTAREALTEST -v -fuzztime 10s -fuzz=FuzzStateSyscallBrk ./mipsevm
 	go test -run NOTAREALTEST -v -fuzztime 10s -fuzz=FuzzStateSyscallClone ./mipsevm

cannon/cmd/run.go

@@ -94,9 +94,6 @@ type Proof struct {
 	OracleKey    hexutil.Bytes `json:"oracle-key,omitempty"`
 	OracleValue  hexutil.Bytes `json:"oracle-value,omitempty"`
 	OracleOffset uint32        `json:"oracle-offset,omitempty"`
-
-	StepInput   hexutil.Bytes `json:"step-input"`
-	OracleInput hexutil.Bytes `json:"oracle-input"`
 }
 
 type rawHint string
@@ -348,14 +345,8 @@ func Run(ctx *cli.Context) error {
 				Post:      postStateHash,
 				StateData: witness.State,
 				ProofData: witness.MemProof,
-				StepInput: witness.EncodeStepInput(),
 			}
 			if witness.HasPreimage() {
-				inp, err := witness.EncodePreimageOracleInput()
-				if err != nil {
-					return fmt.Errorf("failed to encode pre-image oracle input: %w", err)
-				}
-				proof.OracleInput = inp
 				proof.OracleKey = witness.PreimageKey[:]
 				proof.OracleValue = witness.PreimageValue
 				proof.OracleOffset = witness.PreimageOffset

cannon/mipsevm/evm.go

@@ -16,7 +16,6 @@ import (
 	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/core/vm"
-	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/params"
 
 	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
@@ -24,11 +23,26 @@ import (
 )
 
 var (
-	StepBytes4                      = crypto.Keccak256([]byte("step(bytes,bytes)"))[:4]
-	LoadKeccak256PreimagePartBytes4 = crypto.Keccak256([]byte("loadKeccak256PreimagePart(uint256,bytes)"))[:4]
-	LoadLocalDataBytes4             = crypto.Keccak256([]byte("loadLocalData(uint256,bytes32,uint256,uint256)"))[:4]
+	StepBytes4                      []byte
+	LoadKeccak256PreimagePartBytes4 []byte
+	LoadLocalDataBytes4             []byte
 )
 
+func init() {
+	mipsAbi, err := bindings.MIPSMetaData.GetAbi()
+	if err != nil {
+		panic(fmt.Errorf("failed to load MIPS ABI: %w", err))
+	}
+	StepBytes4 = mipsAbi.Methods["step"].ID[:4]
+
+	preimageAbi, err := bindings.PreimageOracleMetaData.GetAbi()
+	if err != nil {
+		panic(fmt.Errorf("failed to load pre-image oracle ABI: %w", err))
+	}
+	LoadKeccak256PreimagePartBytes4 = preimageAbi.Methods["loadKeccak256PreimagePart"].ID[:4]
+	LoadLocalDataBytes4 = preimageAbi.Methods["loadLocalData"].ID[:4]
+}
+
 // LoadContracts loads the Cannon contracts, from op-bindings package
 func LoadContracts() (*Contracts, error) {
 	var mips, oracle Contract

cannon/mipsevm/evm_test.go

@@ -76,13 +76,13 @@ func (m *MIPSEVM) Step(t *testing.T, stepWitness *StepWitness) []byte {
 
 	if stepWitness.HasPreimage() {
 		t.Logf("reading preimage key %x at offset %d", stepWitness.PreimageKey, stepWitness.PreimageOffset)
-		poInput, err := stepWitness.EncodePreimageOracleInput()
+		poInput, err := stepWitness.EncodePreimageOracleInput(0)
 		require.NoError(t, err, "encode preimage oracle input")
 		_, leftOverGas, err := m.env.Call(vm.AccountRef(sender), m.addrs.Oracle, poInput, startingGas, big.NewInt(0))
 		require.NoErrorf(t, err, "evm should not fail, took %d gas", startingGas-leftOverGas)
 	}
 
-	input := stepWitness.EncodeStepInput()
+	input := stepWitness.EncodeStepInput(0)
 	ret, leftOverGas, err := m.env.Call(vm.AccountRef(sender), m.addrs.MIPS, input, startingGas, big.NewInt(0))
 	require.NoError(t, err, "evm should not fail")
 	require.Len(t, ret, 32, "expecting 32-byte state hash")
@@ -241,7 +241,7 @@ func TestEVMFault(t *testing.T) {
 				State:    initialState.EncodeWitness(),
 				MemProof: insnProof[:],
 			}
-			input := stepWitness.EncodeStepInput()
+			input := stepWitness.EncodeStepInput(0)
 			startingGas := uint64(30_000_000)
 
 			_, _, err := env.Call(vm.AccountRef(sender), addrs.MIPS, input, startingGas, big.NewInt(0))

cannon/mipsevm/witness.go

@@ -8,6 +8,8 @@ import (
 	preimage "github.com/ethereum-optimism/optimism/op-preimage"
 )
 
+type LocalContext uint64
+
 type StepWitness struct {
 	// encoded state witness
 	State []byte
@@ -25,7 +27,13 @@ func uint32ToBytes32(v uint32) []byte {
 	return out[:]
 }
 
-func (wit *StepWitness) EncodeStepInput() []byte {
+func uint64ToBytes32(v uint64) []byte {
+	var out [32]byte
+	binary.BigEndian.PutUint64(out[32-8:], v)
+	return out[:]
+}
+
+func (wit *StepWitness) EncodeStepInput(localContext LocalContext) []byte {
 	abiStateLen := len(wit.State)
 	if abiStateLen%32 != 0 {
 		abiStateLen += 32 - (abiStateLen % 32)
@@ -36,8 +44,9 @@ func (wit *StepWitness) EncodeStepInput() []byte {
 
 	var input []byte
 	input = append(input, StepBytes4...)
-	input = append(input, uint32ToBytes32(32*2)...)                          // state data offset in bytes
-	input = append(input, uint32ToBytes32(32*2+32+uint32(len(abiState)))...) // proof data offset in bytes
+	input = append(input, uint32ToBytes32(32*3)...)                          // state data offset in bytes
+	input = append(input, uint32ToBytes32(32*3+32+uint32(len(abiState)))...) // proof data offset in bytes
+	input = append(input, uint64ToBytes32(uint64(localContext))...)          // local context in bytes
 
 	input = append(input, uint32ToBytes32(uint32(len(wit.State)))...) // state data length in bytes
 	input = append(input, abiState[:]...)
@@ -50,7 +59,7 @@ func (wit *StepWitness) HasPreimage() bool {
 	return wit.PreimageKey != ([32]byte{})
 }
 
-func (wit *StepWitness) EncodePreimageOracleInput() ([]byte, error) {
+func (wit *StepWitness) EncodePreimageOracleInput(localContext LocalContext) ([]byte, error) {
 	if wit.PreimageKey == ([32]byte{}) {
 		return nil, errors.New("cannot encode pre-image oracle input, witness has no pre-image to proof")
 	}
@@ -63,6 +72,7 @@ func (wit *StepWitness) EncodePreimageOracleInput() ([]byte, error) {
 		var input []byte
 		input = append(input, LoadLocalDataBytes4...)
 		input = append(input, wit.PreimageKey[:]...)
+		input = append(input, uint64ToBytes32(uint64(localContext))...) // local context in bytes
 
 		preimagePart := wit.PreimageValue[8:]
 		var tmp [32]byte

codecov.yml

-comment: false
+codecov:
+  require_ci_to_pass: false
+comment:
+  layout: "diff, flags, files"
+  behavior: default
+  require_changes: true
+  flags:
+    - contracts-bedrock-tests
+
 ignore:
   - "op-e2e"
-  - "**/*.t.sol"
   - "op-bindings/bindings/*.go"
+  - "**/*.t.sol"
+  - "packages/contracts-bedrock/test/**/*.sol"
+  - "packages/contracts-bedrock/scripts/**/*.sol"
   - "packages/contracts-bedrock/contracts/vendor/WETH9.sol"
   - 'packages/contracts-bedrock/contracts/EAS/**/*.sol'
 coverage:
@@ -13,6 +23,7 @@ coverage:
         threshold: 0% # coverage is not allowed to reduce vs. the PR base
         base: auto
         informational: true
+        enabled: true
     project:
       default:
         informational: true
@@ -22,7 +33,7 @@ flag_management:
   individual_flags:
     - name: contracts-bedrock-tests
       paths:
-        - packages/contracts-bedrock/contracts
+        - packages/contracts-bedrock
       statuses:
         - type: patch
           target: 100%

docker-bake.hcl

+variable "REGISTRY" {
+  default = "us-docker.pkg.dev"
+}
+
+variable "REPOSITORY" {
+  default = "oplabs-tools-artifacts/images"
+}
+
+variable "GIT_COMMIT" {
+  default = "dev"
+}
+
+variable "GIT_DATE" {
+  default = "0"
+}
+
+variable "GIT_VERSION" {
+  default = "docker"  // original default as set in proxyd file, not used by full go stack, yet
+}
+
+variable "IMAGE_TAGS" {
+  default = "${GIT_COMMIT}" // split by ","
+}
+
+variable "PLATFORMS" {
+  // You can override this as "linux/amd64,linux/arm64".
+  // Only a specify a single platform when `--load` ing into docker.
+  // Multi-platform is supported when outputting to disk or pushing to a registry.
+  // Multi-platform builds can be tested locally with:  --set="*.output=type=image,push=false"
+  default = "linux/amd64"
+}
+
+target "op-stack-go" {
+  dockerfile = "ops/docker/op-stack-go/Dockerfile"
+  context = "."
+  args = {
+    GIT_COMMIT = "${GIT_COMMIT}"
+    GIT_DATE = "${GIT_DATE}"
+  }
+  platforms = split(",", PLATFORMS)
+  tags = [for tag in split(",", IMAGE_TAGS) : "${REGISTRY}/${REPOSITORY}/op-stack-go:${tag}"]
+}
+
+target "op-node" {
+  dockerfile = "Dockerfile"
+  context = "./op-node"
+  args = {
+    OP_STACK_GO_BUILDER = "op-stack-go"
+  }
+  contexts = {
+    op-stack-go: "target:op-stack-go"
+  }
+  platforms = split(",", PLATFORMS)
+  tags = [for tag in split(",", IMAGE_TAGS) : "${REGISTRY}/${REPOSITORY}/op-node:${tag}"]
+}
+
+target "op-batcher" {
+  dockerfile = "Dockerfile"
+  context = "./op-batcher"
+  args = {
+    OP_STACK_GO_BUILDER = "op-stack-go"
+  }
+  contexts = {
+    op-stack-go: "target:op-stack-go"
+  }
+  platforms = split(",", PLATFORMS)
+  tags = [for tag in split(",", IMAGE_TAGS) : "${REGISTRY}/${REPOSITORY}/op-batcher:${tag}"]
+}
+
+target "op-proposer" {
+  dockerfile = "Dockerfile"
+  context = "./op-proposer"
+  args = {
+    OP_STACK_GO_BUILDER = "op-stack-go"
+  }
+  contexts = {
+    op-stack-go: "target:op-stack-go"
+  }
+  platforms = split(",", PLATFORMS)
+  tags = [for tag in split(",", IMAGE_TAGS) : "${REGISTRY}/${REPOSITORY}/op-proposer:${tag}"]
+}
+
+target "op-challenger" {
+  dockerfile = "Dockerfile"
+  context = "./op-challenger"
+  args = {
+    OP_STACK_GO_BUILDER = "op-stack-go"
+  }
+  contexts = {
+    op-stack-go: "target:op-stack-go"
+  }
+  platforms = split(",", PLATFORMS)
+  tags = [for tag in split(",", IMAGE_TAGS) : "${REGISTRY}/${REPOSITORY}/op-challenger:${tag}"]
+}
+
+target "op-heartbeat" {
+  dockerfile = "Dockerfile"
+  context = "./op-heartbeat"
+  args = {
+    OP_STACK_GO_BUILDER = "op-stack-go"
+  }
+  contexts = {
+    op-stack-go: "target:op-stack-go"
+  }
+  platforms = split(",", PLATFORMS)
+  tags = [for tag in split(",", IMAGE_TAGS) : "${REGISTRY}/${REPOSITORY}/op-heartbeat:${tag}"]
+}
+
+target "op-program" {
+  dockerfile = "Dockerfile"
+  context = "./op-program"
+  args = {
+    OP_STACK_GO_BUILDER = "op-stack-go"
+  }
+  contexts = {
+    op-stack-go: "target:op-stack-go"
+  }
+  platforms = split(",", PLATFORMS)
+  tags = [for tag in split(",", IMAGE_TAGS) : "${REGISTRY}/${REPOSITORY}/op-program:${tag}"]
+}
+
+target "proxyd" {
+  dockerfile = "Dockerfile"
+  context = "./proxyd"
+  args = {
+    // proxyd dockerfile has no _ in the args
+    GITCOMMIT = "${GIT_COMMIT}"
+    GITDATE = "${GIT_DATE}"
+    GITVERSION = "${GIT_VERSION}"
+  }
+  platforms = split(",", PLATFORMS)
+  tags = [for tag in split(",", IMAGE_TAGS) : "${REGISTRY}/${REPOSITORY}/proxyd:${tag}"]
+}
+
+target "indexer" {
+  dockerfile = "./indexer/Dockerfile"
+  context = "./"
+  args = {
+    // proxyd dockerfile has no _ in the args
+    GITCOMMIT = "${GIT_COMMIT}"
+    GITDATE = "${GIT_DATE}"
+    GITVERSION = "${GIT_VERSION}"
+  }
+  platforms = split(",", PLATFORMS)
+  tags = [for tag in split(",", IMAGE_TAGS) : "${REGISTRY}/${REPOSITORY}/indexer:${tag}"]
+}
+
+target "ufm-metamask" {
+  dockerfile = "Dockerfile"
+  context = "./ufm-test-services/metamask"
+  args = {
+    // proxyd dockerfile has no _ in the args
+    GITCOMMIT = "${GIT_COMMIT}"
+    GITDATE = "${GIT_DATE}"
+    GITVERSION = "${GIT_VERSION}"
+  }
+  platforms = split(",", PLATFORMS)
+  tags = [for tag in split(",", IMAGE_TAGS) : "${REGISTRY}/${REPOSITORY}/ufm-metamask:${tag}"]
+}
+
+target "chain-mon" {
+  dockerfile = "./ops/docker/Dockerfile.packages"
+  context = "."
+  args = {
+    // proxyd dockerfile has no _ in the args
+    GITCOMMIT = "${GIT_COMMIT}"
+    GITDATE = "${GIT_DATE}"
+    GITVERSION = "${GIT_VERSION}"
+  }
+  // this is a multi-stage build, where each stage is a possible output target, but wd-mon is all we publish
+  target = "wd-mon"
+  platforms = split(",", PLATFORMS)
+  tags = [for tag in split(",", IMAGE_TAGS) : "${REGISTRY}/${REPOSITORY}/chain-mon:${tag}"]
+}
+
+target "ci-builder" {
+  dockerfile = "./ops/docker/ci-builder/Dockerfile"
+  context = "."
+  platforms = split(",", PLATFORMS)
+  tags = [for tag in split(",", IMAGE_TAGS) : "${REGISTRY}/${REPOSITORY}/ci-builder:${tag}"]
+}
+
+

docs/fault-proof-alpha/immunefi.md

@@ -6,6 +6,9 @@ During this early phase of ongoing development, we invite security researchers a
 The current system is not production ready, however the core infrastructure for creating an instruction trace ([Cannon][cannon] + the [`op-program`][op-program]), the off-chain challenge agent ([`op-challenger`][op-challenger]),
 and the on-chain infrastructure for the [Dispute Game][dispute-game] are all in place.
 
+For the Fault Proof Alpha security review, we've pinned `546fb2c7a5796b7fe50b0b7edc7666d3bd281d6f` as the commit hash in the monorepo. This commit hash was the head of the `develop` branch at the time of the alpha's launch. All
+security reviews and PoCs should be derived from this commit hash, as the contracts and off-chain agents are being updated frequently at this stage of development.
+
 ### Resources
 
 > **Note**

docs/postmortems/2022-02-02-inflation-vuln.md

@@ -229,7 +229,7 @@ of checking email at the start of their day. This caused some delay in the initi
 
 Early in the process, the existence of the issue was openly discussed in a public slack channel,
 although the details of the vulnerability and exploit path were not described. This violates the
-[principle of least priviledge](https://en.wikipedia.org/wiki/Principle_of_least_privilege), as well
+[principle of least privilege](https://en.wikipedia.org/wiki/Principle_of_least_privilege), as well
 as our already existing incident response protocols
 
 **Action taken:**

endpoint-monitor/Makefile

-GITCOMMIT := $(shell git rev-parse HEAD)
-GITDATE := $(shell git show -s --format='%ct')
+GITCOMMIT ?= $(shell git rev-parse HEAD)
+GITDATE ?= $(shell git show -s --format='%ct')
 VERSION := v0.0.0
 
 LDFLAGSSTRING +=-X main.GitCommit=$(GITCOMMIT)

endpoint-monitor/cmd/main.go

 package main
 
 import (
-	"fmt"
 	"os"
 
+	opservice "github.com/ethereum-optimism/optimism/op-service"
 	oplog "github.com/ethereum-optimism/optimism/op-service/log"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/urfave/cli/v2"
@@ -22,7 +22,7 @@ func main() {
 
 	app := cli.NewApp()
 	app.Flags = endpointMonitor.CLIFlags("ENDPOINT_MONITOR")
-	app.Version = fmt.Sprintf("%s-%s-%s", Version, GitCommit, GitDate)
+	app.Version = opservice.FormatVersion(Version, GitCommit, GitDate, "")
 	app.Name = "endpoint-monitor"
 	app.Usage = "Endpoint Monitoring Service"
 	app.Description = ""

go.mod

@@ -8,32 +8,32 @@ require (
 	github.com/btcsuite/btcd/chaincfg/chainhash v1.0.1
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0
 	github.com/ethereum-optimism/go-ethereum-hdwallet v0.1.3
-	github.com/ethereum-optimism/superchain-registry/superchain v0.0.0-20231001123245-7b48d3818686
+	github.com/ethereum-optimism/superchain-registry/superchain v0.0.0-20231030223232-e16eae11e492
 	github.com/ethereum/go-ethereum v1.13.1
-	github.com/fsnotify/fsnotify v1.6.0
+	github.com/fsnotify/fsnotify v1.7.0
 	github.com/go-chi/chi/v5 v5.0.10
 	github.com/go-chi/docgen v1.2.0
 	github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb
 	github.com/google/go-cmp v0.6.0
 	github.com/google/gofuzz v1.2.1-0.20220503160820-4a35382e8fc8
-	github.com/google/uuid v1.3.1
+	github.com/google/uuid v1.4.0
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/golang-lru/v2 v2.0.5
 	github.com/holiman/uint256 v1.2.3
 	github.com/ipfs/go-datastore v0.6.0
 	github.com/ipfs/go-ds-leveldb v0.5.0
 	github.com/jackc/pgtype v1.14.0
-	github.com/jackc/pgx/v5 v5.4.3
+	github.com/jackc/pgx/v5 v5.5.0
 	github.com/libp2p/go-libp2p v0.31.0
 	github.com/libp2p/go-libp2p-mplex v0.9.0
 	github.com/libp2p/go-libp2p-pubsub v0.9.3
 	github.com/libp2p/go-libp2p-testing v0.12.0
-	github.com/mattn/go-isatty v0.0.19
+	github.com/mattn/go-isatty v0.0.20
 	github.com/multiformats/go-base32 v0.1.0
 	github.com/multiformats/go-multiaddr v0.12.0
 	github.com/multiformats/go-multiaddr-dns v0.3.1
 	github.com/olekukonko/tablewriter v0.0.5
-	github.com/onsi/gomega v1.28.0
+	github.com/onsi/gomega v1.29.0
 	github.com/pkg/errors v0.9.1
 	github.com/pkg/profile v1.7.0
 	github.com/prometheus/client_golang v1.17.0
@@ -41,17 +41,17 @@ require (
 	github.com/urfave/cli/v2 v2.25.7
 	golang.org/x/crypto v0.14.0
 	golang.org/x/exp v0.0.0-20230905200255-921286631fa9
-	golang.org/x/sync v0.4.0
+	golang.org/x/sync v0.5.0
 	golang.org/x/term v0.13.0
-	golang.org/x/time v0.3.0
-	gorm.io/driver/postgres v1.5.3
+	golang.org/x/time v0.4.0
+	gorm.io/driver/postgres v1.5.4
 	gorm.io/gorm v1.25.5
 )
 
 require (
 	github.com/DataDog/zstd v1.5.2 // indirect
 	github.com/Microsoft/go-winio v0.6.1 // indirect
-	github.com/VictoriaMetrics/fastcache v1.10.0 // indirect
+	github.com/VictoriaMetrics/fastcache v1.12.1 // indirect
 	github.com/allegro/bigcache v1.2.1 // indirect
 	github.com/benbjohnson/clock v1.3.5 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
@@ -60,12 +60,13 @@ require (
 	github.com/btcsuite/btcd/btcutil v1.1.0 // indirect
 	github.com/btcsuite/btclog v0.0.0-20170628155309-84c8d2346e9f // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
-	github.com/cockroachdb/errors v1.9.1 // indirect
+	github.com/cockroachdb/errors v1.11.1 // indirect
 	github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b // indirect
-	github.com/cockroachdb/pebble v0.0.0-20230906160148-46873a6a7a06 // indirect
-	github.com/cockroachdb/redact v1.1.3 // indirect
+	github.com/cockroachdb/pebble v0.0.0-20231018212520-f6cde3fc2fa4 // indirect
+	github.com/cockroachdb/redact v1.1.5 // indirect
+	github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 // indirect
 	github.com/consensys/bavard v0.1.13 // indirect
-	github.com/consensys/gnark-crypto v0.12.0 // indirect
+	github.com/consensys/gnark-crypto v0.12.1 // indirect
 	github.com/containerd/cgroups v1.1.0 // indirect
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
@@ -81,7 +82,7 @@ require (
 	github.com/dop251/goja v0.0.0-20230806174421-c933cf95e127 // indirect
 	github.com/elastic/gosigar v0.14.2 // indirect
 	github.com/ethereum/c-kzg-4844 v0.3.1 // indirect
-	github.com/fatih/color v1.7.0 // indirect
+	github.com/fatih/color v1.13.0 // indirect
 	github.com/felixge/fgprof v0.9.3 // indirect
 	github.com/fjl/memsize v0.0.1 // indirect
 	github.com/flynn/noise v1.0.0 // indirect
@@ -96,7 +97,7 @@ require (
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
 	github.com/gofrs/flock v0.8.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
-	github.com/golang-jwt/jwt/v4 v4.4.2 // indirect
+	github.com/golang-jwt/jwt/v4 v4.5.0 // indirect
 	github.com/golang/mock v1.6.0 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/google/gopacket v1.1.19 // indirect
@@ -117,6 +118,7 @@ require (
 	github.com/jackc/pgio v1.0.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jackc/puddle/v2 v2.2.1 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
 	github.com/jbenet/goprocess v0.1.4 // indirect
@@ -161,7 +163,7 @@ require (
 	github.com/multiformats/go-varint v0.0.7 // indirect
 	github.com/naoina/go-stringutil v0.1.0 // indirect
 	github.com/naoina/toml v0.1.2-0.20170918210437-9fafd6967416 // indirect
-	github.com/onsi/ginkgo/v2 v2.12.0 // indirect
+	github.com/onsi/ginkgo/v2 v2.13.0 // indirect
 	github.com/opencontainers/runtime-spec v1.1.0 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 // indirect
@@ -208,7 +210,7 @@ require (
 	rsc.io/tmplfunc v0.0.3 // indirect
 )
 
-replace github.com/ethereum/go-ethereum v1.13.1 => github.com/ethereum-optimism/op-geth v1.101301.0-rc.2.0.20231002141926-1e6910b91798
+replace github.com/ethereum/go-ethereum v1.13.1 => github.com/ethereum-optimism/op-geth v1.101304.0-rc.2.0.20231030225546-cd491fa3b588
 
 //replace github.com/ethereum-optimism/superchain-registry/superchain => ../superchain-registry/superchain
 //replace github.com/ethereum/go-ethereum v1.13.1 => ../go-ethereum

indexer/api-ts/generated.ts

@@ -34,12 +34,12 @@ export interface WithdrawalItem {
   from: string;
   to: string;
   transactionHash: string;
-  messageHash: string;
+  crossDomainMessageHash: string;
   timestamp: number /* uint64 */;
   l2BlockHash: string;
   amount: string;
-  proofTransactionHash: string;
-  claimTransactionHash: string;
+  l1ProvenTxHash: string;
+  l1FinalizedTxHash: string;
   l1TokenAddress: string;
   l2TokenAddress: string;
 }

indexer/api/api.go

@@ -6,33 +6,25 @@ import (
 	"fmt"
 	"net"
 	"net/http"
-	"runtime/debug"
 	"strconv"
-	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/go-chi/chi/v5/middleware"
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/ethereum/go-ethereum/log"
 
 	"github.com/ethereum-optimism/optimism/indexer/api/routes"
 	"github.com/ethereum-optimism/optimism/indexer/config"
 	"github.com/ethereum-optimism/optimism/indexer/database"
 	"github.com/ethereum-optimism/optimism/op-service/httputil"
 	"github.com/ethereum-optimism/optimism/op-service/metrics"
-	"github.com/ethereum/go-ethereum/log"
-	"github.com/go-chi/chi/v5"
-	"github.com/go-chi/chi/v5/middleware"
-	"github.com/prometheus/client_golang/prometheus"
 )
 
 const ethereumAddressRegex = `^0x[a-fA-F0-9]{40}$`
 
-// Api ... Indexer API struct
-// TODO : Structured error responses
-type API struct {
-	log             log.Logger
-	router          *chi.Mux
-	serverConfig    config.ServerConfig
-	metricsConfig   config.ServerConfig
-	metricsRegistry *prometheus.Registry
-}
-
 const (
 	MetricsNamespace = "op_indexer_api"
 	addressParam     = "{address:%s}"
@@ -47,6 +39,23 @@ const (
 	SupplyPath = "/api/v0/supply"
 )
 
+// Api ... Indexer API struct
+// TODO : Structured error responses
+type APIService struct {
+	log    log.Logger
+	router *chi.Mux
+
+	bv      database.BridgeTransfersView
+	dbClose func() error
+
+	metricsRegistry *prometheus.Registry
+
+	apiServer     *httputil.HTTPServer
+	metricsServer *httputil.HTTPServer
+
+	stopped atomic.Bool
+}
+
 // chiMetricsMiddleware ... Injects a metrics recorder into request processing middleware
 func chiMetricsMiddleware(rec metrics.HTTPRecorder) func(http.Handler) http.Handler {
 	return func(next http.Handler) http.Handler {
@@ -55,114 +64,116 @@ func chiMetricsMiddleware(rec metrics.HTTPRecorder) func(http.Handler) http.Hand
 }
 
 // NewApi ... Construct a new api instance
-func NewApi(logger log.Logger, bv database.BridgeTransfersView, serverConfig config.ServerConfig, metricsConfig config.ServerConfig) *API {
-	// (1) Initialize dependencies
-	apiRouter := chi.NewRouter()
-	h := routes.NewRoutes(logger, bv, apiRouter)
-
-	mr := metrics.NewRegistry()
-	promRecorder := metrics.NewPromHTTPRecorder(mr, MetricsNamespace)
-
-	// (2) Inject routing middleware
-	apiRouter.Use(chiMetricsMiddleware(promRecorder))
-	apiRouter.Use(middleware.Recoverer)
-	apiRouter.Use(middleware.Heartbeat(HealthPath))
-
-	// (3) Set GET routes
-	apiRouter.Get(fmt.Sprintf(DepositsPath+addressParam, ethereumAddressRegex), h.L1DepositsHandler)
-	apiRouter.Get(fmt.Sprintf(WithdrawalsPath+addressParam, ethereumAddressRegex), h.L2WithdrawalsHandler)
-
-	apiRouter.Get(SupplyPath, h.SupplyView)
-
-	return &API{log: logger, router: apiRouter, metricsRegistry: mr, serverConfig: serverConfig, metricsConfig: metricsConfig}
+func NewApi(ctx context.Context, log log.Logger, cfg *Config) (*APIService, error) {
+	out := &APIService{log: log, metricsRegistry: metrics.NewRegistry()}
+	if err := out.initFromConfig(ctx, cfg); err != nil {
+		return nil, errors.Join(err, out.Stop(ctx)) // close any resources we may have opened already
+	}
+	return out, nil
 }
 
-// Run ... Runs the API server routines
-func (a *API) Run(ctx context.Context) error {
-	var wg sync.WaitGroup
-	errCh := make(chan error, 2)
-
-	// (1) Construct an inner function that will start a goroutine
-	//    and handle any panics that occur on a shared error channel
-	processCtx, processCancel := context.WithCancel(ctx)
-	runProcess := func(start func(ctx context.Context) error) {
-		wg.Add(1)
-		go func() {
-			defer func() {
-				if err := recover(); err != nil {
-					a.log.Error("halting api on panic", "err", err)
-					debug.PrintStack()
-					errCh <- fmt.Errorf("panic: %v", err)
+func (a *APIService) initFromConfig(ctx context.Context, cfg *Config) error {
+	if err := a.initDB(ctx, cfg.DB); err != nil {
+		return fmt.Errorf("failed to init DB: %w", err)
+	}
+	if err := a.startMetricsServer(cfg.MetricsServer); err != nil {
+		return fmt.Errorf("failed to start metrics server: %w", err)
+	}
+	a.initRouter(cfg.HTTPServer)
+	if err := a.startServer(cfg.HTTPServer); err != nil {
+		return fmt.Errorf("failed to start API server: %w", err)
 	}
+	return nil
+}
 
-				processCancel()
-				wg.Done()
-			}()
+func (a *APIService) Start(ctx context.Context) error {
+	// Completed all setup-up jobs at init-time already,
+	// and the API service does not have any other special starting routines or background-jobs to start.
+	return nil
+}
 
-			errCh <- start(processCtx)
-		}()
+func (a *APIService) Stop(ctx context.Context) error {
+	var result error
+	if a.apiServer != nil {
+		if err := a.apiServer.Stop(ctx); err != nil {
+			result = errors.Join(result, fmt.Errorf("failed to stop API server: %w", err))
+		}
 	}
+	if a.metricsServer != nil {
+		if err := a.metricsServer.Stop(ctx); err != nil {
+			result = errors.Join(result, fmt.Errorf("failed to stop metrics server: %w", err))
+		}
+	}
+	if a.dbClose != nil {
+		if err := a.dbClose(); err != nil {
+			result = errors.Join(result, fmt.Errorf("failed to close DB: %w", err))
+		}
+	}
+	a.stopped.Store(true)
+	a.log.Info("API service shutdown complete")
+	return result
+}
 
-	// (2) Start the API and metrics servers
-	runProcess(a.startServer)
-	runProcess(a.startMetricsServer)
+func (a *APIService) Stopped() bool {
+	return a.stopped.Load()
+}
 
-	// (3) Wait for all processes to complete
-	wg.Wait()
+// Addr ... returns the address that the HTTP server is listening on (excl. http:// prefix, just the host and port)
+func (a *APIService) Addr() string {
+	if a.apiServer == nil {
+		return ""
+	}
+	return a.apiServer.Addr().String()
+}
 
-	err := <-errCh
+func (a *APIService) initDB(ctx context.Context, connector DBConnector) error {
+	db, err := connector.OpenDB(ctx, a.log)
 	if err != nil {
-		a.log.Error("api stopped", "err", err)
-	} else {
-		a.log.Info("api stopped")
+		return fmt.Errorf("failed to connect to databse: %w", err)
 	}
-
-	return err
+	a.dbClose = db.Closer
+	a.bv = db.BridgeTransfers
+	return nil
 }
 
-// Port ... Returns the the port that server is listening on
-func (a *API) Port() int {
-	return a.serverConfig.Port
+func (a *APIService) initRouter(apiConfig config.ServerConfig) {
+	apiRouter := chi.NewRouter()
+	h := routes.NewRoutes(a.log, a.bv, apiRouter)
+
+	promRecorder := metrics.NewPromHTTPRecorder(a.metricsRegistry, MetricsNamespace)
+
+	apiRouter.Use(chiMetricsMiddleware(promRecorder))
+	apiRouter.Use(middleware.Timeout(time.Duration(apiConfig.WriteTimeout) * time.Second))
+	apiRouter.Use(middleware.Recoverer)
+	apiRouter.Use(middleware.Heartbeat(HealthPath))
+
+	apiRouter.Get(fmt.Sprintf(DepositsPath+addressParam, ethereumAddressRegex), h.L1DepositsHandler)
+	apiRouter.Get(fmt.Sprintf(WithdrawalsPath+addressParam, ethereumAddressRegex), h.L2WithdrawalsHandler)
+	apiRouter.Get(SupplyPath, h.SupplyView)
+	a.router = apiRouter
 }
 
 // startServer ... Starts the API server
-func (a *API) startServer(ctx context.Context) error {
-	a.log.Debug("api server listening...", "port", a.serverConfig.Port)
-	addr := net.JoinHostPort(a.serverConfig.Host, strconv.Itoa(a.serverConfig.Port))
+func (a *APIService) startServer(serverConfig config.ServerConfig) error {
+	a.log.Debug("API server listening...", "port", serverConfig.Port)
+	addr := net.JoinHostPort(serverConfig.Host, strconv.Itoa(serverConfig.Port))
 	srv, err := httputil.StartHTTPServer(addr, a.router)
 	if err != nil {
 		return fmt.Errorf("failed to start API server: %w", err)
 	}
-
-	host, portStr, err := net.SplitHostPort(srv.Addr().String())
-	if err != nil {
-		return errors.Join(err, srv.Close())
-	}
-	port, err := strconv.Atoi(portStr)
-	if err != nil {
-		return errors.Join(err, srv.Close())
-	}
-
-	// Update the port in the config in case the OS chose a different port
-	// than the one we requested (e.g. using port 0 to fetch a random open port)
-	a.serverConfig.Host = host
-	a.serverConfig.Port = port
-
-	<-ctx.Done()
-	if err := srv.Stop(context.Background()); err != nil {
-		return fmt.Errorf("failed to shutdown api server: %w", err)
-	}
+	a.log.Info("API server started", "addr", srv.Addr().String())
+	a.apiServer = srv
 	return nil
 }
 
 // startMetricsServer ... Starts the metrics server
-func (a *API) startMetricsServer(ctx context.Context) error {
-	a.log.Debug("starting metrics server...", "port", a.metricsConfig.Port)
-	srv, err := metrics.StartServer(a.metricsRegistry, a.metricsConfig.Host, a.metricsConfig.Port)
+func (a *APIService) startMetricsServer(metricsConfig config.ServerConfig) error {
+	a.log.Debug("starting metrics server...", "port", metricsConfig.Port)
+	srv, err := metrics.StartServer(a.metricsRegistry, metricsConfig.Host, metricsConfig.Port)
 	if err != nil {
 		return fmt.Errorf("failed to start metrics server: %w", err)
 	}
-	<-ctx.Done()
-	defer a.log.Info("metrics server stopped")
-	return srv.Stop(context.Background())
+	a.log.Info("Metrics server started", "addr", srv.Addr().String())
+	a.metricsServer = srv
+	return nil
 }

indexer/api/api_test.go

 package api
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -24,11 +25,12 @@ var mockAddress = "0x4204204204204204204204204204204204204204"
 
 var apiConfig = config.ServerConfig{
 	Host: "localhost",
-	Port: 8080,
+	Port: 0, // random port, to allow parallel tests
 }
+
 var metricsConfig = config.ServerConfig{
 	Host: "localhost",
-	Port: 7300,
+	Port: 0, // random port, to allow parallel tests
 }
 
 var (
@@ -100,8 +102,14 @@ func (mbv *MockBridgeTransfersView) L1BridgeDepositSum() (float64, error) {
 
 func TestHealthz(t *testing.T) {
 	logger := testlog.Logger(t, log.LvlInfo)
-	api := NewApi(logger, &MockBridgeTransfersView{}, apiConfig, metricsConfig)
-	request, err := http.NewRequest("GET", "/healthz", nil)
+	cfg := &Config{
+		DB:            &TestDBConnector{BridgeTransfers: &MockBridgeTransfersView{}},
+		HTTPServer:    apiConfig,
+		MetricsServer: metricsConfig,
+	}
+	api, err := NewApi(context.Background(), logger, cfg)
+	require.NoError(t, err)
+	request, err := http.NewRequest("GET", "http://"+api.Addr()+"/healthz", nil)
 	assert.Nil(t, err)
 
 	responseRecorder := httptest.NewRecorder()
@@ -112,8 +120,14 @@ func TestHealthz(t *testing.T) {
 
 func TestL1BridgeDepositsHandler(t *testing.T) {
 	logger := testlog.Logger(t, log.LvlInfo)
-	api := NewApi(logger, &MockBridgeTransfersView{}, apiConfig, metricsConfig)
-	request, err := http.NewRequest("GET", fmt.Sprintf("/api/v0/deposits/%s", mockAddress), nil)
+	cfg := &Config{
+		DB:            &TestDBConnector{BridgeTransfers: &MockBridgeTransfersView{}},
+		HTTPServer:    apiConfig,
+		MetricsServer: metricsConfig,
+	}
+	api, err := NewApi(context.Background(), logger, cfg)
+	require.NoError(t, err)
+	request, err := http.NewRequest("GET", fmt.Sprintf("http://"+api.Addr()+"/api/v0/deposits/%s", mockAddress), nil)
 	assert.Nil(t, err)
 
 	responseRecorder := httptest.NewRecorder()
@@ -135,8 +149,14 @@ func TestL1BridgeDepositsHandler(t *testing.T) {
 
 func TestL2BridgeWithdrawalsByAddressHandler(t *testing.T) {
 	logger := testlog.Logger(t, log.LvlInfo)
-	api := NewApi(logger, &MockBridgeTransfersView{}, apiConfig, metricsConfig)
-	request, err := http.NewRequest("GET", fmt.Sprintf("/api/v0/withdrawals/%s", mockAddress), nil)
+	cfg := &Config{
+		DB:            &TestDBConnector{BridgeTransfers: &MockBridgeTransfersView{}},
+		HTTPServer:    apiConfig,
+		MetricsServer: metricsConfig,
+	}
+	api, err := NewApi(context.Background(), logger, cfg)
+	require.NoError(t, err)
+	request, err := http.NewRequest("GET", fmt.Sprintf("http://"+api.Addr()+"/api/v0/withdrawals/%s", mockAddress), nil)
 	assert.Nil(t, err)
 
 	responseRecorder := httptest.NewRecorder()
@@ -154,8 +174,8 @@ func TestL2BridgeWithdrawalsByAddressHandler(t *testing.T) {
 	assert.Equal(t, resp.Items[0].To, withdrawal.Tx.ToAddress.String())
 	assert.Equal(t, resp.Items[0].TransactionHash, common.HexToHash("0x789").String())
 	assert.Equal(t, resp.Items[0].Amount, withdrawal.Tx.Amount.String())
-	assert.Equal(t, resp.Items[0].ProofTransactionHash, common.HexToHash("0x123").String())
-	assert.Equal(t, resp.Items[0].ClaimTransactionHash, common.HexToHash("0x123").String())
+	assert.Equal(t, resp.Items[0].L1ProvenTxHash, common.HexToHash("0x123").String())
+	assert.Equal(t, resp.Items[0].L1FinalizedTxHash, common.HexToHash("0x123").String())
 	assert.Equal(t, resp.Items[0].L1TokenAddress, withdrawal.TokenPair.RemoteTokenAddress.String())
 	assert.Equal(t, resp.Items[0].L2TokenAddress, withdrawal.TokenPair.LocalTokenAddress.String())
 	assert.Equal(t, resp.Items[0].Timestamp, withdrawal.Tx.Timestamp)

indexer/api/config.go

+package api
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/log"
+
+	"github.com/ethereum-optimism/optimism/indexer/config"
+	"github.com/ethereum-optimism/optimism/indexer/database"
+)
+
+// DB represents the abstract DB access the API has.
+type DB struct {
+	BridgeTransfers database.BridgeTransfersView
+	Closer          func() error
+}
+
+// DBConfigConnector implements a fully config based DBConnector
+type DBConfigConnector struct {
+	config.DBConfig
+}
+
+func (cfg *DBConfigConnector) OpenDB(ctx context.Context, log log.Logger) (*DB, error) {
+	db, err := database.NewDB(ctx, log, cfg.DBConfig)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to databse: %w", err)
+	}
+	return &DB{
+		BridgeTransfers: db.BridgeTransfers,
+		Closer:          db.Close,
+	}, nil
+}
+
+type TestDBConnector struct {
+	BridgeTransfers database.BridgeTransfersView
+}
+
+func (tdb *TestDBConnector) OpenDB(ctx context.Context, log log.Logger) (*DB, error) {
+	return &DB{
+		BridgeTransfers: tdb.BridgeTransfers,
+		Closer: func() error {
+			log.Info("API service closed test DB view")
+			return nil
+		},
+	}, nil
+}
+
+// DBConnector is an interface: the config may provide different ways to access the DB.
+// This is implemented in tests to provide custom DB views, or share the DB with other services.
+type DBConnector interface {
+	OpenDB(ctx context.Context, log log.Logger) (*DB, error)
+}
+
+// Config for the API service
+type Config struct {
+	DB            DBConnector
+	HTTPServer    config.ServerConfig
+	MetricsServer config.ServerConfig
+}

indexer/api/models/models.go

 package models
 
+import (
+	"github.com/ethereum-optimism/optimism/indexer/database"
+	"github.com/ethereum/go-ethereum/common"
+)
+
 // DepositItem ... Deposit item model for API responses
 type DepositItem struct {
 	Guid           string `json:"guid"`
@@ -27,12 +32,12 @@ type WithdrawalItem struct {
 	From                   string `json:"from"`
 	To                     string `json:"to"`
 	TransactionHash        string `json:"transactionHash"`
-	MessageHash          string `json:"messageHash"`
+	CrossDomainMessageHash string `json:"crossDomainMessageHash"`
 	Timestamp              uint64 `json:"timestamp"`
 	L2BlockHash            string `json:"l2BlockHash"`
 	Amount                 string `json:"amount"`
-	ProofTransactionHash string `json:"proofTransactionHash"`
-	ClaimTransactionHash string `json:"claimTransactionHash"`
+	L1ProvenTxHash         string `json:"l1ProvenTxHash"`
+	L1FinalizedTxHash      string `json:"l1FinalizedTxHash"`
 	L1TokenAddress         string `json:"l1TokenAddress"`
 	L2TokenAddress         string `json:"l2TokenAddress"`
 }
@@ -47,3 +52,38 @@ type WithdrawalResponse struct {
 type BridgeSupplyView struct {
 	L1DepositSum float64 `json:"l1DepositSum"`
 }
+
+// FIXME make a pure function that returns a struct instead of newWithdrawalResponse
+// newWithdrawalResponse ... Converts a database.L2BridgeWithdrawalsResponse to an api.WithdrawalResponse
+func CreateWithdrawalResponse(withdrawals *database.L2BridgeWithdrawalsResponse) WithdrawalResponse {
+	items := make([]WithdrawalItem, len(withdrawals.Withdrawals))
+	for i, withdrawal := range withdrawals.Withdrawals {
+
+		cdh := withdrawal.L2BridgeWithdrawal.CrossDomainMessageHash
+		if cdh == nil { // Zero value indicates that the withdrawal didn't have a cross domain message
+			cdh = &common.Hash{0}
+		}
+
+		item := WithdrawalItem{
+			Guid:                   withdrawal.L2BridgeWithdrawal.TransactionWithdrawalHash.String(),
+			L2BlockHash:            withdrawal.L2BlockHash.String(),
+			Timestamp:              withdrawal.L2BridgeWithdrawal.Tx.Timestamp,
+			From:                   withdrawal.L2BridgeWithdrawal.Tx.FromAddress.String(),
+			To:                     withdrawal.L2BridgeWithdrawal.Tx.ToAddress.String(),
+			TransactionHash:        withdrawal.L2TransactionHash.String(),
+			Amount:                 withdrawal.L2BridgeWithdrawal.Tx.Amount.String(),
+			CrossDomainMessageHash: cdh.String(),
+			L1ProvenTxHash:         withdrawal.ProvenL1TransactionHash.String(),
+			L1FinalizedTxHash:      withdrawal.FinalizedL1TransactionHash.String(),
+			L1TokenAddress:         withdrawal.L2BridgeWithdrawal.TokenPair.RemoteTokenAddress.String(),
+			L2TokenAddress:         withdrawal.L2BridgeWithdrawal.TokenPair.LocalTokenAddress.String(),
+		}
+		items[i] = item
+	}
+
+	return WithdrawalResponse{
+		Cursor:      withdrawals.Cursor,
+		HasNextPage: withdrawals.HasNextPage,
+		Items:       items,
+	}
+}

indexer/api/models/models_test.go

+package models_test
+
+import (
+	"fmt"
+	"reflect"
+	"testing"
+
+	"github.com/ethereum-optimism/optimism/indexer/api/models"
+	"github.com/ethereum-optimism/optimism/indexer/database"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCreateWithdrawal(t *testing.T) {
+	// (1) Create a dummy database response object
+
+	cdh := common.HexToHash("0x2")
+	dbWithdrawals := &database.L2BridgeWithdrawalsResponse{
+		Withdrawals: []database.L2BridgeWithdrawalWithTransactionHashes{
+			{
+				L2BridgeWithdrawal: database.L2BridgeWithdrawal{
+					TransactionWithdrawalHash: common.HexToHash("0x1"),
+					BridgeTransfer: database.BridgeTransfer{
+						CrossDomainMessageHash: &cdh,
+						Tx: database.Transaction{
+							FromAddress: common.HexToAddress("0x3"),
+							ToAddress:   common.HexToAddress("0x4"),
+							Timestamp:   5,
+						},
+						TokenPair: database.TokenPair{
+							LocalTokenAddress:  common.HexToAddress("0x6"),
+							RemoteTokenAddress: common.HexToAddress("0x7"),
+						},
+					},
+				},
+			},
+		},
+	}
+
+	// (2) Create and validate response object
+
+	response := models.CreateWithdrawalResponse(dbWithdrawals)
+	require.NotEmpty(t, response.Items)
+	require.Len(t, response.Items, 1)
+
+	// (3) Use reflection to check that all fields in WithdrawalItem are populated correctly
+
+	item := response.Items[0]
+	structType := reflect.TypeOf(item)
+
+	structVal := reflect.ValueOf(item)
+	fieldNum := structVal.NumField()
+
+	for i := 0; i < fieldNum; i++ {
+		field := structVal.Field(i)
+		fieldName := structType.Field(i).Name
+
+		isSet := field.IsValid() && !field.IsZero()
+
+		require.True(t, isSet, fmt.Sprintf("%s in not set", fieldName))
+
+	}
+
+}

indexer/api/routes/withdrawals.go

@@ -4,37 +4,9 @@ import (
 	"net/http"
 
 	"github.com/ethereum-optimism/optimism/indexer/api/models"
-	"github.com/ethereum-optimism/optimism/indexer/database"
 	"github.com/go-chi/chi/v5"
 )
 
-// FIXME make a pure function that returns a struct instead of newWithdrawalResponse
-// newWithdrawalResponse ... Converts a database.L2BridgeWithdrawalsResponse to an api.WithdrawalResponse
-func newWithdrawalResponse(withdrawals *database.L2BridgeWithdrawalsResponse) models.WithdrawalResponse {
-	items := make([]models.WithdrawalItem, len(withdrawals.Withdrawals))
-	for i, withdrawal := range withdrawals.Withdrawals {
-		item := models.WithdrawalItem{
-			Guid:                 withdrawal.L2BridgeWithdrawal.TransactionWithdrawalHash.String(),
-			L2BlockHash:          withdrawal.L2BlockHash.String(),
-			From:                 withdrawal.L2BridgeWithdrawal.Tx.FromAddress.String(),
-			To:                   withdrawal.L2BridgeWithdrawal.Tx.ToAddress.String(),
-			TransactionHash:      withdrawal.L2TransactionHash.String(),
-			Amount:               withdrawal.L2BridgeWithdrawal.Tx.Amount.String(),
-			ProofTransactionHash: withdrawal.ProvenL1TransactionHash.String(),
-			ClaimTransactionHash: withdrawal.FinalizedL1TransactionHash.String(),
-			L1TokenAddress:       withdrawal.L2BridgeWithdrawal.TokenPair.RemoteTokenAddress.String(),
-			L2TokenAddress:       withdrawal.L2BridgeWithdrawal.TokenPair.LocalTokenAddress.String(),
-		}
-		items[i] = item
-	}
-
-	return models.WithdrawalResponse{
-		Cursor:      withdrawals.Cursor,
-		HasNextPage: withdrawals.HasNextPage,
-		Items:       items,
-	}
-}
-
 // L2WithdrawalsHandler ... Handles /api/v0/withdrawals/{address} GET requests
 func (h Routes) L2WithdrawalsHandler(w http.ResponseWriter, r *http.Request) {
 	addressValue := chi.URLParam(r, "address")
@@ -68,7 +40,7 @@ func (h Routes) L2WithdrawalsHandler(w http.ResponseWriter, r *http.Request) {
 		h.logger.Error("Unable to read withdrawals from DB", "err", err.Error())
 		return
 	}
-	response := newWithdrawalResponse(withdrawals)
+	response := models.CreateWithdrawalResponse(withdrawals)
 
 	err = jsonResponse(w, response, http.StatusOK)
 	if err != nil {

indexer/bigint/bigint.go

@@ -25,3 +25,9 @@ func Clamp(start, end *big.Int, size uint64) *big.Int {
 func Matcher(num int64) func(*big.Int) bool {
 	return func(bi *big.Int) bool { return bi.Int64() == num }
 }
+
+func WeiToETH(wei *big.Int) *big.Float {
+	f := new(big.Float)
+	f.SetString(wei.String())
+	return f.Quo(f, big.NewFloat(1e18))
+}

indexer/cmd/indexer/cli.go

 package main
 
 import (
+	"context"
+
+	"github.com/urfave/cli/v2"
+
+	"github.com/ethereum/go-ethereum/params"
+
 	"github.com/ethereum-optimism/optimism/indexer"
 	"github.com/ethereum-optimism/optimism/indexer/api"
 	"github.com/ethereum-optimism/optimism/indexer/config"
 	"github.com/ethereum-optimism/optimism/indexer/database"
+	"github.com/ethereum-optimism/optimism/op-service/cliapp"
 	oplog "github.com/ethereum-optimism/optimism/op-service/log"
-	"github.com/ethereum/go-ethereum/params"
-
-	"github.com/urfave/cli/v2"
+	"github.com/ethereum-optimism/optimism/op-service/opio"
 )
 
 var (
@@ -27,68 +32,62 @@ var (
 	}
 )
 
-func runIndexer(ctx *cli.Context) error {
+func runIndexer(ctx *cli.Context, shutdown context.CancelCauseFunc) (cliapp.Lifecycle, error) {
 	log := oplog.NewLogger(oplog.AppOut(ctx), oplog.ReadCLIConfig(ctx)).New("role", "indexer")
 	oplog.SetGlobalLogHandler(log.GetHandler())
+	log.Info("running indexer...")
+
 	cfg, err := config.LoadConfig(log, ctx.String(ConfigFlag.Name))
 	if err != nil {
 		log.Error("failed to load config", "err", err)
-		return err
-	}
-
-	db, err := database.NewDB(log, cfg.DB)
-	if err != nil {
-		log.Error("failed to connect to database", "err", err)
-		return err
+		return nil, err
 	}
-	defer db.Close()
 
-	indexer, err := indexer.NewIndexer(log, db, cfg.Chain, cfg.RPCs, cfg.HTTPServer, cfg.MetricsServer)
-	if err != nil {
-		log.Error("failed to create indexer", "err", err)
-		return err
-	}
-
-	return indexer.Run(ctx.Context)
+	return indexer.NewIndexer(ctx.Context, log, &cfg, shutdown)
 }
 
-func runApi(ctx *cli.Context) error {
+func runApi(ctx *cli.Context, _ context.CancelCauseFunc) (cliapp.Lifecycle, error) {
 	log := oplog.NewLogger(oplog.AppOut(ctx), oplog.ReadCLIConfig(ctx)).New("role", "api")
 	oplog.SetGlobalLogHandler(log.GetHandler())
+	log.Info("running api...")
+
 	cfg, err := config.LoadConfig(log, ctx.String(ConfigFlag.Name))
 	if err != nil {
 		log.Error("failed to load config", "err", err)
-		return err
+		return nil, err
 	}
 
-	db, err := database.NewDB(log, cfg.DB)
-	if err != nil {
-		log.Error("failed to connect to database", "err", err)
-		return err
+	apiCfg := &api.Config{
+		DB:            &api.DBConfigConnector{DBConfig: cfg.DB},
+		HTTPServer:    cfg.HTTPServer,
+		MetricsServer: cfg.MetricsServer,
 	}
-	defer db.Close()
 
-	api := api.NewApi(log, db.BridgeTransfers, cfg.HTTPServer, cfg.MetricsServer)
-	return api.Run(ctx.Context)
+	return api.NewApi(ctx.Context, log, apiCfg)
 }
 
 func runMigrations(ctx *cli.Context) error {
-	log := oplog.NewLogger(oplog.AppOut(ctx), oplog.ReadCLIConfig(ctx)).New("role", "api")
+	// We don't maintain a complicated lifecycle here, just interrupt to shut down.
+	ctx.Context = opio.CancelOnInterrupt(ctx.Context)
+
+	log := oplog.NewLogger(oplog.AppOut(ctx), oplog.ReadCLIConfig(ctx)).New("role", "migrations")
 	oplog.SetGlobalLogHandler(log.GetHandler())
+	log.Info("running migrations...")
+
 	cfg, err := config.LoadConfig(log, ctx.String(ConfigFlag.Name))
-	migrationsDir := ctx.String(MigrationsFlag.Name)
 	if err != nil {
 		log.Error("failed to load config", "err", err)
 		return err
 	}
 
-	db, err := database.NewDB(log, cfg.DB)
+	db, err := database.NewDB(ctx.Context, log, cfg.DB)
 	if err != nil {
 		log.Error("failed to connect to database", "err", err)
 		return err
 	}
 	defer db.Close()
 
+	migrationsDir := ctx.String(MigrationsFlag.Name)
 	return db.ExecuteSQLMigration(migrationsDir)
 }
 
@@ -106,13 +105,13 @@ func newCli(GitCommit string, GitDate string) *cli.App {
 				Name:        "api",
 				Flags:       flags,
 				Description: "Runs the api service",
-				Action:      runApi,
+				Action:      cliapp.LifecycleCmd(runApi),
 			},
 			{
 				Name:        "index",
 				Flags:       flags,
 				Description: "Runs the indexing service",
-				Action:      runIndexer,
+				Action:      cliapp.LifecycleCmd(runIndexer),
 			},
 			{
 				Name:        "migrate",

indexer/cmd/indexer/main.go

@@ -4,9 +4,10 @@ import (
 	"context"
 	"os"
 
+	"github.com/ethereum/go-ethereum/log"
+
 	oplog "github.com/ethereum-optimism/optimism/op-service/log"
 	"github.com/ethereum-optimism/optimism/op-service/opio"
-	"github.com/ethereum/go-ethereum/log"
 )
 
 var (
@@ -15,17 +16,12 @@ var (
 )
 
 func main() {
-	// This is the most root context, used to propagate
-	// cancellations to all spawned application-level goroutines
-	ctx, cancel := context.WithCancel(context.Background())
-	go func() {
-		opio.BlockOnInterrupts()
-		cancel()
-	}()
-
 	oplog.SetupDefaults()
 	app := newCli(GitCommit, GitDate)
+	// sub-commands set up their individual interrupt lifecycles, which can block on the given interrupt as needed.
+	ctx := opio.WithInterruptBlocker(context.Background())
 	if err := app.RunContext(ctx, os.Args); err != nil {
 		log.Error("application failed", "err", err)
+		os.Exit(1)
 	}
 }

indexer/config/config.go

@@ -134,10 +134,11 @@ type DBConfig struct {
 	Password string `toml:"password"`
 }
 
-// Configures the a server
+// Configures the server
 type ServerConfig struct {
 	Host         string `toml:"host"`
 	Port         int    `toml:"port"`
+	WriteTimeout int    `toml:"timeout"`
 }
 
 // LoadConfig loads the `indexer.toml` config file from a given path
@@ -153,18 +154,25 @@ func LoadConfig(log log.Logger, path string) (Config, error) {
 	data = []byte(os.ExpandEnv(string(data)))
 	log.Debug("parsed config file", "data", string(data))
 
-	if _, err := toml.Decode(string(data), &cfg); err != nil {
+	md, err := toml.Decode(string(data), &cfg)
+	if err != nil {
 		log.Error("failed to decode config file", "err", err)
 		return cfg, err
 	}
 
+	if len(md.Undecoded()) > 0 {
+		log.Error("unknown fields in config file", "fields", md.Undecoded())
+		err = fmt.Errorf("unknown fields in config file: %v", md.Undecoded())
+		return cfg, err
+	}
+
 	if cfg.Chain.Preset == DevnetPresetId {
 		preset, err := DevnetPreset()
 		if err != nil {
 			return cfg, err
 		}
 
-		log.Info("loaded local devnet preset")
+		log.Info("detected preset", "preset", DevnetPresetId, "name", preset.Name)
 		cfg.Chain = preset.ChainConfig
 	} else if cfg.Chain.Preset != 0 {
 		preset, ok := Presets[cfg.Chain.Preset]
@@ -191,25 +199,21 @@ func LoadConfig(log log.Logger, path string) (Config, error) {
 	// Defaults for any unset options
 
 	if cfg.Chain.L1PollingInterval == 0 {
-		log.Info("setting default L1 polling interval", "interval", defaultLoopInterval)
 		cfg.Chain.L1PollingInterval = defaultLoopInterval
 	}
 
 	if cfg.Chain.L2PollingInterval == 0 {
-		log.Info("setting default L2 polling interval", "interval", defaultLoopInterval)
 		cfg.Chain.L2PollingInterval = defaultLoopInterval
 	}
 
 	if cfg.Chain.L1HeaderBufferSize == 0 {
-		log.Info("setting default L1 header buffer", "size", defaultHeaderBufferSize)
 		cfg.Chain.L1HeaderBufferSize = defaultHeaderBufferSize
 	}
 
 	if cfg.Chain.L2HeaderBufferSize == 0 {
-		log.Info("setting default L2 header buffer", "size", defaultHeaderBufferSize)
 		cfg.Chain.L2HeaderBufferSize = defaultHeaderBufferSize
 	}
 
-	log.Info("loaded config", "config", cfg.Chain)
+	log.Info("loaded chain config", "config", cfg.Chain)
 	return cfg, nil
 }

indexer/config/config_test.go

@@ -257,3 +257,49 @@ func TestLocalDevnet(t *testing.T) {
 
 	require.Equal(t, devnetPreset.ChainConfig.L1Contracts, conf.Chain.L1Contracts)
 }
+
+func TestThrowsOnUnknownKeys(t *testing.T) {
+	logger := testlog.Logger(t, log.LvlInfo)
+	tmpfile, err := os.CreateTemp("", "test.toml")
+	require.NoError(t, err)
+	defer os.Remove(tmpfile.Name())
+	defer tmpfile.Close()
+
+	testData := `
+		[chain]
+    unknown_key = 420
+		preset = 420
+
+		[rpcs]
+		l1-rpc = "https://l1.example.com"
+		l2-rpc = "https://l2.example.com"
+
+		[db]
+	  another_unknownKey = 420
+		host = "127.0.0.1"
+		port = 5432
+		user = "postgres"
+		password = "postgres"
+	    name = "indexer"
+
+		[http]
+		host = "127.0.0.1"
+		port = 8080
+
+		[metrics]
+		host = "127.0.0.1"
+		port = 7300
+	`
+
+	data := []byte(testData)
+	err = os.WriteFile(tmpfile.Name(), data, 0644)
+	require.NoError(t, err)
+	defer os.Remove(tmpfile.Name())
+
+	err = tmpfile.Close()
+	require.NoError(t, err)
+
+	_, err = LoadConfig(logger, tmpfile.Name())
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "unknown fields in config file")
+}

indexer/database/blocks.go

@@ -7,8 +7,10 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
 
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )
 
 /**
@@ -67,17 +69,23 @@ type BlocksDB interface {
  */
 
 type blocksDB struct {
+	log  log.Logger
 	gorm *gorm.DB
 }
 
-func newBlocksDB(db *gorm.DB) BlocksDB {
-	return &blocksDB{gorm: db}
+func newBlocksDB(log log.Logger, db *gorm.DB) BlocksDB {
+	return &blocksDB{log: log.New("table", "blocks"), gorm: db}
 }
 
 // L1
 
 func (db *blocksDB) StoreL1BlockHeaders(headers []L1BlockHeader) error {
-	result := db.gorm.CreateInBatches(&headers, batchInsertSize)
+	deduped := db.gorm.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "hash"}}, DoNothing: true})
+	result := deduped.Create(&headers)
+	if result.Error == nil && int(result.RowsAffected) < len(headers) {
+		db.log.Warn("ignored L1 block duplicates", "duplicates", len(headers)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 
@@ -115,7 +123,12 @@ func (db *blocksDB) L1LatestBlockHeader() (*L1BlockHeader, error) {
 // L2
 
 func (db *blocksDB) StoreL2BlockHeaders(headers []L2BlockHeader) error {
-	result := db.gorm.CreateInBatches(&headers, batchInsertSize)
+	deduped := db.gorm.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "hash"}}, DoNothing: true})
+	result := deduped.Create(&headers)
+	if result.Error == nil && int(result.RowsAffected) < len(headers) {
+		db.log.Warn("ignored L2 block duplicates", "duplicates", len(headers)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 
@@ -173,7 +186,6 @@ func (db *blocksDB) LatestObservedEpoch(fromL1Height *big.Int, maxL1Range uint64
 	var header L1BlockHeader
 	if fromL1Height != nil {
 		result := db.gorm.Where("number = ?", fromL1Height).Take(&header)
-		// TODO - Embed logging to db
 		if result.Error != nil {
 			if errors.Is(result.Error, gorm.ErrRecordNotFound) {
 				return nil, nil
@@ -183,7 +195,8 @@ func (db *blocksDB) LatestObservedEpoch(fromL1Height *big.Int, maxL1Range uint64
 
 		fromTimestamp = header.Timestamp
 	} else {
-		result := db.gorm.Order("number desc").Take(&header)
+		// Take the lowest indexed L1 block to compute the lower bound
+		result := db.gorm.Order("number ASC").Take(&header)
 		if result.Error != nil {
 			if errors.Is(result.Error, gorm.ErrRecordNotFound) {
 				return nil, nil
@@ -192,6 +205,7 @@ func (db *blocksDB) LatestObservedEpoch(fromL1Height *big.Int, maxL1Range uint64
 		}
 
 		fromL1Height = header.Number
+		fromTimestamp = header.Timestamp
 	}
 
 	// Upper Bound (lowest timestamp indexed between L1/L2 bounded by `maxL1Range`)

indexer/database/bridge_messages.go

@@ -6,8 +6,10 @@ import (
 	"math/big"
 
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/log"
 
 	"github.com/google/uuid"
 )
@@ -60,11 +62,12 @@ type BridgeMessagesDB interface {
  */
 
 type bridgeMessagesDB struct {
+	log  log.Logger
 	gorm *gorm.DB
 }
 
-func newBridgeMessagesDB(db *gorm.DB) BridgeMessagesDB {
-	return &bridgeMessagesDB{gorm: db}
+func newBridgeMessagesDB(log log.Logger, db *gorm.DB) BridgeMessagesDB {
+	return &bridgeMessagesDB{log: log.New("table", "bridge_messages"), gorm: db}
 }
 
 /**
@@ -72,7 +75,12 @@ func newBridgeMessagesDB(db *gorm.DB) BridgeMessagesDB {
  */
 
 func (db bridgeMessagesDB) StoreL1BridgeMessages(messages []L1BridgeMessage) error {
-	result := db.gorm.CreateInBatches(&messages, batchInsertSize)
+	deduped := db.gorm.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "message_hash"}}, DoNothing: true})
+	result := deduped.Create(&messages)
+	if result.Error == nil && int(result.RowsAffected) < len(messages) {
+		db.log.Warn("ignored L1 bridge message duplicates", "duplicates", len(messages)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 
@@ -98,7 +106,13 @@ func (db bridgeMessagesDB) MarkRelayedL1BridgeMessage(messageHash common.Hash, r
 	if err != nil {
 		return err
 	} else if message == nil {
-		return fmt.Errorf("L1BridgeMessage with message hash %s not found", messageHash)
+		return fmt.Errorf("L1BridgeMessage %s not found", messageHash)
+	}
+
+	if message.RelayedMessageEventGUID != nil && message.RelayedMessageEventGUID.ID() == relayEvent.ID() {
+		return nil
+	} else if message.RelayedMessageEventGUID != nil {
+		return fmt.Errorf("relayed message %s re-relayed with a different event %d", messageHash, relayEvent)
 	}
 
 	message.RelayedMessageEventGUID = &relayEvent
@@ -111,7 +125,12 @@ func (db bridgeMessagesDB) MarkRelayedL1BridgeMessage(messageHash common.Hash, r
  */
 
 func (db bridgeMessagesDB) StoreL2BridgeMessages(messages []L2BridgeMessage) error {
-	result := db.gorm.CreateInBatches(&messages, batchInsertSize)
+	deduped := db.gorm.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "message_hash"}}, DoNothing: true})
+	result := deduped.Create(&messages)
+	if result.Error == nil && int(result.RowsAffected) < len(messages) {
+		db.log.Warn("ignored L2 bridge message duplicates", "duplicates", len(messages)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 
@@ -137,7 +156,13 @@ func (db bridgeMessagesDB) MarkRelayedL2BridgeMessage(messageHash common.Hash, r
 	if err != nil {
 		return err
 	} else if message == nil {
-		return fmt.Errorf("L2BridgeMessage with message hash %s not found", messageHash)
+		return fmt.Errorf("L2BridgeMessage %s not found", messageHash)
+	}
+
+	if message.RelayedMessageEventGUID != nil && message.RelayedMessageEventGUID.ID() == relayEvent.ID() {
+		return nil
+	} else if message.RelayedMessageEventGUID != nil {
+		return fmt.Errorf("relayed message %s re-relayed with a different event %s", messageHash, relayEvent)
 	}
 
 	message.RelayedMessageEventGUID = &relayEvent

indexer/database/bridge_transactions.go

@@ -7,8 +7,10 @@ import (
 
 	"github.com/google/uuid"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/log"
 )
 
 /**
@@ -68,11 +70,12 @@ type BridgeTransactionsDB interface {
  */
 
 type bridgeTransactionsDB struct {
+	log  log.Logger
 	gorm *gorm.DB
 }
 
-func newBridgeTransactionsDB(db *gorm.DB) BridgeTransactionsDB {
-	return &bridgeTransactionsDB{gorm: db}
+func newBridgeTransactionsDB(log log.Logger, db *gorm.DB) BridgeTransactionsDB {
+	return &bridgeTransactionsDB{log: log.New("table", "bridge_transactions"), gorm: db}
 }
 
 /**
@@ -80,7 +83,12 @@ func newBridgeTransactionsDB(db *gorm.DB) BridgeTransactionsDB {
  */
 
 func (db *bridgeTransactionsDB) StoreL1TransactionDeposits(deposits []L1TransactionDeposit) error {
-	result := db.gorm.CreateInBatches(&deposits, batchInsertSize)
+	deduped := db.gorm.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "source_hash"}}, DoNothing: true})
+	result := deduped.Create(&deposits)
+	if result.Error == nil && int(result.RowsAffected) < len(deposits) {
+		db.log.Warn("ignored L1 tx deposit duplicates", "duplicates", len(deposits)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 
@@ -133,7 +141,12 @@ func (db *bridgeTransactionsDB) L1LatestBlockHeader() (*L1BlockHeader, error) {
  */
 
 func (db *bridgeTransactionsDB) StoreL2TransactionWithdrawals(withdrawals []L2TransactionWithdrawal) error {
-	result := db.gorm.CreateInBatches(&withdrawals, batchInsertSize)
+	deduped := db.gorm.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "withdrawal_hash"}}, DoNothing: true})
+	result := deduped.Create(&withdrawals)
+	if result.Error == nil && int(result.RowsAffected) < len(withdrawals) {
+		db.log.Warn("ignored L2 tx withdrawal duplicates", "duplicates", len(withdrawals)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 
@@ -155,11 +168,16 @@ func (db *bridgeTransactionsDB) MarkL2TransactionWithdrawalProvenEvent(withdrawa
 	withdrawal, err := db.L2TransactionWithdrawal(withdrawalHash)
 	if err != nil {
 		return err
-	}
-	if withdrawal == nil {
+	} else if withdrawal == nil {
 		return fmt.Errorf("transaction withdrawal hash %s not found", withdrawalHash)
 	}
 
+	if withdrawal.ProvenL1EventGUID != nil && withdrawal.ProvenL1EventGUID.ID() == provenL1EventGuid.ID() {
+		return nil
+	} else if withdrawal.ProvenL1EventGUID != nil {
+		return fmt.Errorf("proven withdrawal %s re-proven with a different event %s", withdrawalHash, provenL1EventGuid)
+	}
+
 	withdrawal.ProvenL1EventGUID = &provenL1EventGuid
 	result := db.gorm.Save(&withdrawal)
 	return result.Error
@@ -170,14 +188,18 @@ func (db *bridgeTransactionsDB) MarkL2TransactionWithdrawalFinalizedEvent(withdr
 	withdrawal, err := db.L2TransactionWithdrawal(withdrawalHash)
 	if err != nil {
 		return err
-	}
-	if withdrawal == nil {
+	} else if withdrawal == nil {
 		return fmt.Errorf("transaction withdrawal hash %s not found", withdrawalHash)
-	}
-	if withdrawal.ProvenL1EventGUID == nil {
+	} else if withdrawal.ProvenL1EventGUID == nil {
 		return fmt.Errorf("cannot mark unproven withdrawal hash %s as finalized", withdrawal.WithdrawalHash)
 	}
 
+	if withdrawal.FinalizedL1EventGUID != nil && withdrawal.FinalizedL1EventGUID.ID() == finalizedL1EventGuid.ID() {
+		return nil
+	} else if withdrawal.FinalizedL1EventGUID != nil {
+		return fmt.Errorf("finalized withdrawal %s re-finalized with a different event %s", withdrawalHash, finalizedL1EventGuid)
+	}
+
 	withdrawal.FinalizedL1EventGUID = &finalizedL1EventGuid
 	withdrawal.Succeeded = &succeeded
 	result := db.gorm.Save(&withdrawal)

indexer/database/bridge_transfers.go

@@ -5,9 +5,11 @@ import (
 	"fmt"
 
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 
 	"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/log"
 )
 
 var (
@@ -80,11 +82,12 @@ type BridgeTransfersDB interface {
  */
 
 type bridgeTransfersDB struct {
+	log  log.Logger
 	gorm *gorm.DB
 }
 
-func newBridgeTransfersDB(db *gorm.DB) BridgeTransfersDB {
-	return &bridgeTransfersDB{gorm: db}
+func newBridgeTransfersDB(log log.Logger, db *gorm.DB) BridgeTransfersDB {
+	return &bridgeTransfersDB{log: log.New("table", "bridge_transfers"), gorm: db}
 }
 
 /**
@@ -92,7 +95,12 @@ func newBridgeTransfersDB(db *gorm.DB) BridgeTransfersDB {
  */
 
 func (db *bridgeTransfersDB) StoreL1BridgeDeposits(deposits []L1BridgeDeposit) error {
-	result := db.gorm.CreateInBatches(&deposits, batchInsertSize)
+	deduped := db.gorm.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "transaction_source_hash"}}, DoNothing: true})
+	result := deduped.Create(&deposits)
+	if result.Error == nil && int(result.RowsAffected) < len(deposits) {
+		db.log.Warn("ignored L1 bridge transfer duplicates", "duplicates", len(deposits)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 
@@ -159,12 +167,11 @@ func (db *bridgeTransfersDB) L1BridgeDepositsByAddress(address common.Address, c
 		cursorClause = fmt.Sprintf("l1_transaction_deposits.timestamp <= %d", txDeposit.Tx.Timestamp)
 	}
 
-	// TODO join with l1_bridged_tokens and l2_bridged_tokens
 	ethAddressString := predeploys.LegacyERC20ETHAddr.String()
 
 	// Coalesce l1 transaction deposits that are simply ETH sends
 	ethTransactionDeposits := db.gorm.Model(&L1TransactionDeposit{})
-	ethTransactionDeposits = ethTransactionDeposits.Where(&Transaction{FromAddress: address}).Where("data = '0x' AND amount > 0")
+	ethTransactionDeposits = ethTransactionDeposits.Where(&Transaction{FromAddress: address}).Where("amount > 0")
 	ethTransactionDeposits = ethTransactionDeposits.Joins("INNER JOIN l1_contract_events ON l1_contract_events.guid = initiated_l1_event_guid")
 	ethTransactionDeposits = ethTransactionDeposits.Select(`
 from_address, to_address, amount, data, source_hash AS transaction_source_hash,
@@ -217,7 +224,12 @@ l1_bridge_deposits.timestamp, cross_domain_message_hash, local_token_address, re
  */
 
 func (db *bridgeTransfersDB) StoreL2BridgeWithdrawals(withdrawals []L2BridgeWithdrawal) error {
-	result := db.gorm.CreateInBatches(&withdrawals, batchInsertSize)
+	deduped := db.gorm.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "transaction_withdrawal_hash"}}, DoNothing: true})
+	result := deduped.Create(&withdrawals)
+	if result.Error == nil && int(result.RowsAffected) < len(withdrawals) {
+		db.log.Warn("ignored L2 bridge transfer duplicates", "duplicates", len(withdrawals)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 
@@ -283,7 +295,7 @@ func (db *bridgeTransfersDB) L2BridgeWithdrawalsByAddress(address common.Address
 
 	// Coalesce l2 transaction withdrawals that are simply ETH sends
 	ethTransactionWithdrawals := db.gorm.Model(&L2TransactionWithdrawal{})
-	ethTransactionWithdrawals = ethTransactionWithdrawals.Where(&Transaction{FromAddress: address}).Where(`data = '0x' AND amount > 0`)
+	ethTransactionWithdrawals = ethTransactionWithdrawals.Where(&Transaction{FromAddress: address}).Where("amount > 0")
 	ethTransactionWithdrawals = ethTransactionWithdrawals.Joins("INNER JOIN l2_contract_events ON l2_contract_events.guid = l2_transaction_withdrawals.initiated_l2_event_guid")
 	ethTransactionWithdrawals = ethTransactionWithdrawals.Joins("LEFT JOIN l1_contract_events AS proven_l1_events ON proven_l1_events.guid = l2_transaction_withdrawals.proven_l1_event_guid")
 	ethTransactionWithdrawals = ethTransactionWithdrawals.Joins("LEFT JOIN l1_contract_events AS finalized_l1_events ON finalized_l1_events.guid = l2_transaction_withdrawals.finalized_l1_event_guid")

indexer/database/contract_events.go

@@ -6,9 +6,11 @@ import (
 	"math/big"
 
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
 
 	"github.com/google/uuid"
 )
@@ -99,17 +101,25 @@ type ContractEventsDB interface {
  */
 
 type contractEventsDB struct {
+	log  log.Logger
 	gorm *gorm.DB
 }
 
-func newContractEventsDB(db *gorm.DB) ContractEventsDB {
-	return &contractEventsDB{gorm: db}
+func newContractEventsDB(log log.Logger, db *gorm.DB) ContractEventsDB {
+	return &contractEventsDB{log: log.New("table", "events"), gorm: db}
 }
 
 // L1
 
 func (db *contractEventsDB) StoreL1ContractEvents(events []L1ContractEvent) error {
-	result := db.gorm.CreateInBatches(&events, batchInsertSize)
+	// Since the block hash refers back to L1, we dont necessarily have to check
+	// that the RLP bytes match when doing conflict resolution.
+	deduped := db.gorm.Clauses(clause.OnConflict{OnConstraint: "l1_contract_events_block_hash_log_index_key", DoNothing: true})
+	result := deduped.Create(&events)
+	if result.Error == nil && int(result.RowsAffected) < len(events) {
+		db.log.Warn("ignored L1 contract event duplicates", "duplicates", len(events)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 
@@ -144,7 +154,7 @@ func (db *contractEventsDB) L1ContractEventsWithFilter(filter ContractEvent, fro
 	query := db.gorm.Table("l1_contract_events").Where(&filter)
 	query = query.Joins("INNER JOIN l1_block_headers ON l1_contract_events.block_hash = l1_block_headers.hash")
 	query = query.Where("l1_block_headers.number >= ? AND l1_block_headers.number <= ?", fromHeight, toHeight)
-	query = query.Order("l1_block_headers.number ASC").Select("l1_contract_events.*")
+	query = query.Order("l1_block_headers.number ASC, l1_contract_events.log_index ASC").Select("l1_contract_events.*")
 
 	// NOTE: We use `Find` here instead of `Scan` since `Scan` doesn't not support
 	// model hooks like `ContractEvent#AfterFind`. Functionally they are the same
@@ -176,7 +186,14 @@ func (db *contractEventsDB) L1LatestContractEventWithFilter(filter ContractEvent
 // L2
 
 func (db *contractEventsDB) StoreL2ContractEvents(events []L2ContractEvent) error {
-	result := db.gorm.CreateInBatches(&events, batchInsertSize)
+	// Since the block hash refers back to L2, we dont necessarily have to check
+	// that the RLP bytes match when doing conflict resolution.
+	deduped := db.gorm.Clauses(clause.OnConflict{OnConstraint: "l2_contract_events_block_hash_log_index_key", DoNothing: true})
+	result := deduped.Create(&events)
+	if result.Error == nil && int(result.RowsAffected) < len(events) {
+		db.log.Warn("ignored L2 contract event duplicates", "duplicates", len(events)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 
@@ -211,7 +228,7 @@ func (db *contractEventsDB) L2ContractEventsWithFilter(filter ContractEvent, fro
 	query := db.gorm.Table("l2_contract_events").Where(&filter)
 	query = query.Joins("INNER JOIN l2_block_headers ON l2_contract_events.block_hash = l2_block_headers.hash")
 	query = query.Where("l2_block_headers.number >= ? AND l2_block_headers.number <= ?", fromHeight, toHeight)
-	query = query.Order("l2_block_headers.number ASC").Select("l2_contract_events.*")
+	query = query.Order("l2_block_headers.number ASC, l2_contract_events.log_index ASC").Select("l2_contract_events.*")
 
 	// NOTE: We use `Find` here instead of `Scan` since `Scan` doesn't not support
 	// model hooks like `ContractEvent#AfterFind`. Functionally they are the same

indexer/database/db.go

@@ -10,6 +10,7 @@ import (
 	"github.com/ethereum-optimism/optimism/indexer/config"
 	_ "github.com/ethereum-optimism/optimism/indexer/database/serializers"
 	"github.com/ethereum-optimism/optimism/op-service/retry"
+
 	"github.com/pkg/errors"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -18,16 +19,9 @@ import (
 	"gorm.io/gorm"
 )
 
-var (
-	// The postgres parameter counter for a given query is stored via a  uint16,
-	// resulting in a parameter limit of 65535. In order to avoid reaching this limit
-	// we'll utilize a batch size of 3k for inserts, well below as long as the the number
-	// of columns < 20.
-	batchInsertSize int = 3_000
-)
-
 type DB struct {
 	gorm *gorm.DB
+	log  log.Logger
 
 	Blocks             BlocksDB
 	ContractEvents     ContractEventsDB
@@ -36,8 +30,10 @@ type DB struct {
 	BridgeTransactions BridgeTransactionsDB
 }
 
-func NewDB(log log.Logger, dbConfig config.DBConfig) (*DB, error) {
-	retryStrategy := &retry.ExponentialStrategy{Min: 1000, Max: 20_000, MaxJitter: 250}
+// NewDB connects to the configured DB, and provides client-bindings to it.
+// The initial connection may fail, or the dial may be cancelled with the provided context.
+func NewDB(ctx context.Context, log log.Logger, dbConfig config.DBConfig) (*DB, error) {
+	log = log.New("module", "db")
 
 	dsn := fmt.Sprintf("host=%s dbname=%s sslmode=disable", dbConfig.Host, dbConfig.Name)
 	if dbConfig.Port != 0 {
@@ -51,11 +47,19 @@ func NewDB(log log.Logger, dbConfig config.DBConfig) (*DB, error) {
 	}
 
 	gormConfig := gorm.Config{
+		Logger: newLogger(log),
+
 		// The indexer will explicitly manage the transactions
 		SkipDefaultTransaction: true,
-		Logger:                 newLogger(log),
+
+		// The postgres parameter counter for a given query is represented with uint16,
+		// resulting in a parameter limit of 65535. In order to avoid reaching this limit
+		// we'll utilize a batch size of 3k for inserts, well below the limit as long as
+		// the number of columns < 20.
+		CreateBatchSize: 3_000,
 	}
 
+	retryStrategy := &retry.ExponentialStrategy{Min: 1000, Max: 20_000, MaxJitter: 250}
 	gorm, err := retry.Do[*gorm.DB](context.Background(), 10, retryStrategy, func() (*gorm.DB, error) {
 		gorm, err := gorm.Open(postgres.Open(dsn), &gormConfig)
 		if err != nil {
@@ -66,16 +70,17 @@ func NewDB(log log.Logger, dbConfig config.DBConfig) (*DB, error) {
 	})
 
 	if err != nil {
-		return nil, fmt.Errorf("failed to connect to database after multiple retries: %w", err)
+		return nil, err
 	}
 
 	db := &DB{
 		gorm:               gorm,
-		Blocks:             newBlocksDB(gorm),
-		ContractEvents:     newContractEventsDB(gorm),
-		BridgeTransfers:    newBridgeTransfersDB(gorm),
-		BridgeMessages:     newBridgeMessagesDB(gorm),
-		BridgeTransactions: newBridgeTransactionsDB(gorm),
+		log:                log,
+		Blocks:             newBlocksDB(log, gorm),
+		ContractEvents:     newContractEventsDB(log, gorm),
+		BridgeTransfers:    newBridgeTransfersDB(log, gorm),
+		BridgeMessages:     newBridgeMessagesDB(log, gorm),
+		BridgeTransactions: newBridgeTransactionsDB(log, gorm),
 	}
 
 	return db, nil
@@ -85,11 +90,21 @@ func NewDB(log log.Logger, dbConfig config.DBConfig) (*DB, error) {
 // transaction. If the supplied function errors, the transaction is rolled back.
 func (db *DB) Transaction(fn func(db *DB) error) error {
 	return db.gorm.Transaction(func(tx *gorm.DB) error {
-		return fn(dbFromGormTx(tx))
+		txDB := &DB{
+			gorm:               tx,
+			Blocks:             newBlocksDB(db.log, tx),
+			ContractEvents:     newContractEventsDB(db.log, tx),
+			BridgeTransfers:    newBridgeTransfersDB(db.log, tx),
+			BridgeMessages:     newBridgeMessagesDB(db.log, tx),
+			BridgeTransactions: newBridgeTransactionsDB(db.log, tx),
+		}
+
+		return fn(txDB)
 	})
 }
 
 func (db *DB) Close() error {
+	db.log.Info("closing database")
 	sql, err := db.gorm.DB()
 	if err != nil {
 		return err
@@ -98,17 +113,6 @@ func (db *DB) Close() error {
 	return sql.Close()
 }
 
-func dbFromGormTx(tx *gorm.DB) *DB {
-	return &DB{
-		gorm:               tx,
-		Blocks:             newBlocksDB(tx),
-		ContractEvents:     newContractEventsDB(tx),
-		BridgeTransfers:    newBridgeTransfersDB(tx),
-		BridgeMessages:     newBridgeMessagesDB(tx),
-		BridgeTransactions: newBridgeTransactionsDB(tx),
-	}
-}
-
 func (db *DB) ExecuteSQLMigration(migrationsFolder string) error {
 	err := filepath.Walk(migrationsFolder, func(path string, info os.FileInfo, err error) error {
 		// Check for any walking error
@@ -122,12 +126,14 @@ func (db *DB) ExecuteSQLMigration(migrationsFolder string) error {
 		}
 
 		// Read the migration file content
+		db.log.Info("reading sql file", "path", path)
 		fileContent, readErr := os.ReadFile(path)
 		if readErr != nil {
 			return errors.Wrap(readErr, fmt.Sprintf("Error reading SQL file: %s", path))
 		}
 
 		// Execute the migration
+		db.log.Info("executing sql file", "path", path)
 		execErr := db.gorm.Exec(string(fileContent)).Error
 		if execErr != nil {
 			return errors.Wrap(execErr, fmt.Sprintf("Error executing SQL script: %s", path))
@@ -136,5 +142,6 @@ func (db *DB) ExecuteSQLMigration(migrationsFolder string) error {
 		return nil
 	})
 
+	db.log.Info("finished migrations")
 	return err
 }

indexer/database/logger.go

@@ -14,7 +14,7 @@ import (
 var (
 	_ logger.Interface = Logger{}
 
-	SlowThresholdMilliseconds = 200
+	SlowThresholdMilliseconds int64 = 500
 )
 
 type Logger struct {
@@ -22,7 +22,7 @@ type Logger struct {
 }
 
 func newLogger(log log.Logger) Logger {
-	return Logger{log.New("module", "db")}
+	return Logger{log}
 }
 
 func (l Logger) LogMode(lvl logger.LogLevel) logger.Interface {
@@ -50,7 +50,7 @@ func (l Logger) Trace(ctx context.Context, begin time.Time, fc func() (sql strin
 		sql = fmt.Sprintf("%sVALUES (...)", sql[:i])
 	}
 
-	if elapsedMs < 200 {
+	if elapsedMs < SlowThresholdMilliseconds {
 		l.log.Debug("database operation", "duration_ms", elapsedMs, "rows_affected", rows, "sql", sql)
 	} else {
 		l.log.Warn("database operation", "duration_ms", elapsedMs, "rows_affected", rows, "sql", sql)

indexer/database/serializers/bytes.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"strings"
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"gorm.io/gorm/schema"
@@ -70,5 +71,5 @@ func (BytesSerializer) Value(ctx context.Context, field *schema.Field, dst refle
 	}
 
 	hexStr := hexutil.Encode(fieldBytes.Bytes())
-	return hexStr, nil
+	return strings.ToLower(hexStr), nil
 }

indexer/database/serializers/rlp.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"strings"
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/rlp"
@@ -52,5 +53,5 @@ func (RLPSerializer) Value(ctx context.Context, field *schema.Field, dst reflect
 	}
 
 	hexStr := hexutil.Encode(rlpBytes)
-	return hexStr, nil
+	return strings.ToLower(hexStr), nil
 }

indexer/docker-compose.yml

@@ -58,7 +58,6 @@ services:
     depends_on:
       postgres:
         condition: service_healthy
-    depends_on:
       migrations:
         condition: service_started
 
@@ -155,7 +154,7 @@ services:
       - MAINNET_BEDROCK=true
       - TRM_API_KEY=$TRM_API_KEY
       - GOOGLE_CLOUD_STORAGE_BUCKET_NAME=oplabs-dev-web-content
-      # Recommened to uncomment for local dev unless you need it
+      # Recommend to uncomment for local dev unless you need it
       #- BYPASS_EVENT_LOG_POLLER_BOOTSTRAP=true
     ports:
       - 7422:7300

indexer/e2e_tests/setup.go

@@ -34,7 +34,7 @@ type E2ETestSuite struct {
 
 	// API
 	Client *client.Client
-	API    *api.API
+	API    *api.APIService
 
 	// Indexer
 	DB      *database.DB
@@ -73,7 +73,7 @@ func createE2ETestSuite(t *testing.T) E2ETestSuite {
 	t.Cleanup(func() { opSys.Close() })
 
 	// Indexer Configuration and Start
-	indexerCfg := config.Config{
+	indexerCfg := &config.Config{
 		DB: config.DBConfig{
 			Host: "127.0.0.1",
 			Port: 5432,
@@ -106,51 +106,40 @@ func createE2ETestSuite(t *testing.T) E2ETestSuite {
 	// the system is running, mark this test for Parallel execution
 	t.Parallel()
 
-	// provide a DB for the unit test. disable logging
-	silentLog := testlog.Logger(t, log.LvlInfo)
-	silentLog.SetHandler(log.DiscardHandler())
-	db, err := database.NewDB(silentLog, indexerCfg.DB)
-	require.NoError(t, err)
-	t.Cleanup(func() { db.Close() })
-
 	indexerLog := testlog.Logger(t, log.LvlInfo).New("role", "indexer")
-	indexer, err := indexer.NewIndexer(indexerLog, db, indexerCfg.Chain, indexerCfg.RPCs, indexerCfg.HTTPServer, indexerCfg.MetricsServer)
+	ix, err := indexer.NewIndexer(context.Background(), indexerLog, indexerCfg, func(cause error) {
+		if cause != nil {
+			t.Fatalf("indexer shut down with critical error: %v", cause)
+		}
+	})
 	require.NoError(t, err)
 
-	indexerCtx, indexerStop := context.WithCancel(context.Background())
-	go func() {
-		err := indexer.Run(indexerCtx)
-		if err != nil { // panicking here ensures that the test will exit
-			// during service failure. Using t.Fail() wouldn't be caught
-			// until all awaiting routines finish which would never happen.
-			panic(err)
-		}
-	}()
+	require.NoError(t, ix.Start(context.Background()), "cleanly start indexer")
+
+	t.Cleanup(func() {
+		require.NoError(t, ix.Stop(context.Background()), "cleanly shut down indexer")
+	})
 
 	apiLog := testlog.Logger(t, log.LvlInfo).New("role", "indexer_api")
 
-	apiCfg := config.ServerConfig{
+	apiCfg := &api.Config{
+		DB: &api.TestDBConnector{BridgeTransfers: ix.DB.BridgeTransfers}, // reuse the same DB
+		HTTPServer: config.ServerConfig{
 			Host: "127.0.0.1",
 			Port: 0,
-	}
-
-	mCfg := config.ServerConfig{
+		},
+		MetricsServer: config.ServerConfig{
 			Host: "127.0.0.1",
 			Port: 0,
+		},
 	}
 
-	api := api.NewApi(apiLog, db.BridgeTransfers, apiCfg, mCfg)
-	apiCtx, apiStop := context.WithCancel(context.Background())
-	go func() {
-		err := api.Run(apiCtx)
-		if err != nil {
-			panic(err)
-		}
-	}()
+	apiService, err := api.NewApi(context.Background(), apiLog, apiCfg)
+	require.NoError(t, err, "create indexer API service")
 
+	require.NoError(t, apiService.Start(context.Background()), "start indexer API service")
 	t.Cleanup(func() {
-		apiStop()
-		indexerStop()
+		require.NoError(t, apiService.Stop(context.Background()), "cleanly shut down indexer")
 	})
 
 	// Wait for the API to start listening
@@ -158,16 +147,15 @@ func createE2ETestSuite(t *testing.T) E2ETestSuite {
 
 	client, err := client.NewClient(&client.Config{
 		PaginationLimit: 100,
-		BaseURL:         fmt.Sprintf("http://%s:%d", apiCfg.Host, api.Port()),
+		BaseURL:         "http://" + apiService.Addr(),
 	})
-
-	require.NoError(t, err)
+	require.NoError(t, err, "must open indexer API client")
 
 	return E2ETestSuite{
 		t:        t,
 		Client:   client,
-		DB:       db,
-		Indexer:  indexer,
+		DB:       ix.DB,
+		Indexer:  ix,
 		OpCfg:    &opCfg,
 		OpSys:    opSys,
 		L1Client: opSys.Clients["l1"],
@@ -203,7 +191,7 @@ func setupTestDatabase(t *testing.T) string {
 
 	silentLog := log.New()
 	silentLog.SetHandler(log.DiscardHandler())
-	db, err := database.NewDB(silentLog, dbConfig)
+	db, err := database.NewDB(context.Background(), silentLog, dbConfig)
 	require.NoError(t, err)
 	defer db.Close()
 

indexer/etl/etl.go

@@ -3,14 +3,17 @@ package etl
 import (
 	"context"
 	"errors"
+	"fmt"
 	"math/big"
 	"time"
 
-	"github.com/ethereum-optimism/optimism/indexer/node"
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
+
+	"github.com/ethereum-optimism/optimism/indexer/node"
+	"github.com/ethereum-optimism/optimism/op-service/clock"
 )
 
 type Config struct {
@@ -30,9 +33,15 @@ type ETL struct {
 	headerTraversal  *node.HeaderTraversal
 
 	contracts  []common.Address
-	etlBatches chan ETLBatch
+	etlBatches chan *ETLBatch
 
 	EthClient node.EthClient
+
+	// A reference that'll stay populated between intervals
+	// in the event of failures in order to retry.
+	headers []types.Header
+
+	worker *clock.LoopFn
 }
 
 type ETLBatch struct {
@@ -45,47 +54,54 @@ type ETLBatch struct {
 	HeadersWithLog map[common.Hash]bool
 }
 
-func (etl *ETL) Start(ctx context.Context) error {
-	done := ctx.Done()
-	pollTicker := time.NewTicker(etl.loopInterval)
-	defer pollTicker.Stop()
-
-	// A reference that'll stay populated between intervals
-	// in the event of failures in order to retry.
-	var headers []types.Header
-
+// Start starts the ETL polling routine. The ETL work should be stopped with Close().
+func (etl *ETL) Start() error {
+	if etl.worker != nil {
+		return errors.New("already started")
+	}
 	etl.log.Info("starting etl...")
-	for {
-		select {
-		case <-done:
-			etl.log.Info("stopping etl")
+	etl.worker = clock.NewLoopFn(clock.SystemClock, etl.tick, func() error {
+		close(etl.etlBatches) // can close the channel now, to signal to the consumer that we're done
+		etl.log.Info("stopped etl worker loop")
+		return nil
+	}, etl.loopInterval)
 	return nil
+}
 
-		case <-pollTicker.C:
+func (etl *ETL) Close() error {
+	if etl.worker == nil {
+		return nil // worker was not running
+	}
+	return etl.worker.Close()
+}
+
+func (etl *ETL) tick(_ context.Context) {
 	done := etl.metrics.RecordInterval()
-			if len(headers) > 0 {
+	if len(etl.headers) > 0 {
 		etl.log.Info("retrying previous batch")
 	} else {
-				newHeaders, err := etl.headerTraversal.NextFinalizedHeaders(etl.headerBufferSize)
+		newHeaders, err := etl.headerTraversal.NextHeaders(etl.headerBufferSize)
 		if err != nil {
 			etl.log.Error("error querying for headers", "err", err)
 		} else if len(newHeaders) == 0 {
-					etl.log.Warn("no new headers. processor unexpectedly at head...")
+			etl.log.Warn("no new headers. etl at head?")
 		} else {
-					headers = newHeaders
-					etl.metrics.RecordBatchHeaders(len(newHeaders))
+			etl.headers = newHeaders
+		}
+
+		latestHeader := etl.headerTraversal.LatestHeader()
+		if latestHeader != nil {
+			etl.metrics.RecordLatestHeight(latestHeader.Number)
 		}
 	}
 
 	// only clear the reference if we were able to process this batch
-			err := etl.processBatch(headers)
+	err := etl.processBatch(etl.headers)
 	if err == nil {
-				headers = nil
+		etl.headers = nil
 	}
 
 	done(err)
-		}
-	}
 }
 
 func (etl *ETL) processBatch(headers []types.Header) error {
@@ -97,7 +113,6 @@ func (etl *ETL) processBatch(headers []types.Header) error {
 	batchLog := etl.log.New("batch_start_block_number", firstHeader.Number, "batch_end_block_number", lastHeader.Number)
 	batchLog.Info("extracting batch", "size", len(headers))
 
-	etl.metrics.RecordBatchLatestHeight(lastHeader.Number)
 	headerMap := make(map[common.Hash]*types.Header, len(headers))
 	for i := range headers {
 		header := headers[i]
@@ -105,31 +120,40 @@ func (etl *ETL) processBatch(headers []types.Header) error {
 	}
 
 	headersWithLog := make(map[common.Hash]bool, len(headers))
-	logs, err := etl.EthClient.FilterLogs(ethereum.FilterQuery{FromBlock: firstHeader.Number, ToBlock: lastHeader.Number, Addresses: etl.contracts})
+	filterQuery := ethereum.FilterQuery{FromBlock: firstHeader.Number, ToBlock: lastHeader.Number, Addresses: etl.contracts}
+	logs, err := etl.EthClient.FilterLogs(filterQuery)
 	if err != nil {
 		batchLog.Info("failed to extract logs", "err", err)
 		return err
 	}
-	if len(logs) > 0 {
-		batchLog.Info("detected logs", "size", len(logs))
+
+	if logs.ToBlockHeader.Number.Cmp(lastHeader.Number) != 0 {
+		// Warn and simply wait for the provider to synchronize state
+		batchLog.Warn("mismatch in FilterLog#ToBlock number", "queried_to_block_number", lastHeader.Number, "reported_to_block_number", logs.ToBlockHeader.Number)
+		return fmt.Errorf("mismatch in FilterLog#ToBlock number")
+	} else if logs.ToBlockHeader.Hash() != lastHeader.Hash() {
+		batchLog.Error("mismatch in FitlerLog#ToBlock block hash!!!", "queried_to_block_hash", lastHeader.Hash().String(), "reported_to_block_hash", logs.ToBlockHeader.Hash().String())
+		return fmt.Errorf("mismatch in FitlerLog#ToBlock block hash!!!")
+	}
+
+	if len(logs.Logs) > 0 {
+		batchLog.Info("detected logs", "size", len(logs.Logs))
 	}
 
-	for i := range logs {
-		log := logs[i]
+	for i := range logs.Logs {
+		log := logs.Logs[i]
+		headersWithLog[log.BlockHash] = true
 		if _, ok := headerMap[log.BlockHash]; !ok {
 			// NOTE. Definitely an error state if the none of the headers were re-orged out in between
 			// the blocks and logs retrieval operations. Unlikely as long as the confirmation depth has
 			// been appropriately set or when we get to natively handling reorgs.
-			batchLog.Error("log found with block hash not in the batch", "block_hash", logs[i].BlockHash, "log_index", logs[i].Index)
+			batchLog.Error("log found with block hash not in the batch", "block_hash", logs.Logs[i].BlockHash, "log_index", logs.Logs[i].Index)
 			return errors.New("parsed log with a block hash not in the batch")
 		}
-
-		etl.metrics.RecordBatchLog(log.Address)
-		headersWithLog[log.BlockHash] = true
 	}
 
 	// ensure we use unique downstream references for the etl batch
 	headersRef := headers
-	etl.etlBatches <- ETLBatch{Logger: batchLog, Headers: headersRef, HeaderMap: headerMap, Logs: logs, HeadersWithLog: headersWithLog}
+	etl.etlBatches <- &ETLBatch{Logger: batchLog, Headers: headersRef, HeaderMap: headerMap, Logs: logs.Logs, HeadersWithLog: headersWithLog}
 	return nil
 }

indexer/etl/l1_etl.go

@@ -8,26 +8,37 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+
 	"github.com/ethereum-optimism/optimism/indexer/config"
 	"github.com/ethereum-optimism/optimism/indexer/database"
 	"github.com/ethereum-optimism/optimism/indexer/node"
 	"github.com/ethereum-optimism/optimism/op-service/retry"
-	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/core/types"
-	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum-optimism/optimism/op-service/tasks"
 )
 
 type L1ETL struct {
 	ETL
 
+	// the batch handler may do work that we can interrupt on shutdown
+	resourceCtx    context.Context
+	resourceCancel context.CancelFunc
+
+	tasks tasks.Group
+
 	db *database.DB
-	mu        *sync.Mutex
+
+	mu sync.Mutex
+
 	listeners []chan interface{}
 }
 
 // NewL1ETL creates a new L1ETL instance that will start indexing from different starting points
 // depending on the state of the database and the supplied start height.
-func NewL1ETL(cfg Config, log log.Logger, db *database.DB, metrics Metricer, client node.EthClient, contracts config.L1Contracts) (*L1ETL, error) {
+func NewL1ETL(cfg Config, log log.Logger, db *database.DB, metrics Metricer, client node.EthClient,
+	contracts config.L1Contracts, shutdown context.CancelCauseFunc) (*L1ETL, error) {
 	log = log.New("etl", "l1")
 
 	zeroAddr := common.Address{}
@@ -71,8 +82,10 @@ func NewL1ETL(cfg Config, log log.Logger, db *database.DB, metrics Metricer, cli
 	}
 
 	// NOTE - The use of un-buffered channel here assumes that downstream consumers
-	// will be able to keep up with the rate of incoming batches
-	etlBatches := make(chan ETLBatch)
+	// will be able to keep up with the rate of incoming batches.
+	// When the producer closes the channel we stop consuming from it.
+	etlBatches := make(chan *ETLBatch)
+
 	etl := ETL{
 		loopInterval:     time.Duration(cfg.LoopIntervalMsec) * time.Millisecond,
 		headerBufferSize: uint64(cfg.HeaderBufferSize),
@@ -86,42 +99,78 @@ func NewL1ETL(cfg Config, log log.Logger, db *database.DB, metrics Metricer, cli
 		EthClient: client,
 	}
 
-	return &L1ETL{ETL: etl, db: db, mu: new(sync.Mutex)}, nil
+	resCtx, resCancel := context.WithCancel(context.Background())
+	return &L1ETL{
+		ETL:            etl,
+		db:             db,
+		resourceCtx:    resCtx,
+		resourceCancel: resCancel,
+		tasks: tasks.Group{HandleCrit: func(err error) {
+			shutdown(fmt.Errorf("critical error in L1 ETL: %w", err))
+		}},
+	}, nil
 }
 
-func (l1Etl *L1ETL) Start(ctx context.Context) error {
-	errCh := make(chan error, 1)
-	go func() {
-		errCh <- l1Etl.ETL.Start(ctx)
-	}()
+func (l1Etl *L1ETL) Close() error {
+	var result error
+	// close the producer
+	if err := l1Etl.ETL.Close(); err != nil {
+		result = errors.Join(result, fmt.Errorf("failed to close internal ETL: %w", err))
+	}
+	// tell the consumer it can stop what it's doing
+	l1Etl.resourceCancel()
+	// wait for consumer to pick up on closure of producer
+	if err := l1Etl.tasks.Wait(); err != nil {
+		result = errors.Join(result, fmt.Errorf("failed to await batch handler completion: %w", err))
+	}
+	return result
+}
 
+func (l1Etl *L1ETL) Start() error {
+	// start ETL batch producer
+	if err := l1Etl.ETL.Start(); err != nil {
+		return fmt.Errorf("failed to start internal ETL: %w", err)
+	}
+	// start ETL batch consumer
+	l1Etl.tasks.Go(func() error {
 		for {
-		select {
-		case err := <-errCh:
-			return err
-
 			// Index incoming batches (only L1 blocks that have an emitted log)
-		case batch := <-l1Etl.etlBatches:
+			batch, ok := <-l1Etl.etlBatches
+			if !ok {
+				l1Etl.log.Info("No more batches, shutting down L1 batch handler")
+				return nil
+			}
+			if err := l1Etl.handleBatch(batch); err != nil {
+				return fmt.Errorf("failed to handle batch, stopping L2 ETL: %w", err)
+			}
+		}
+	})
+	return nil
+}
+
+func (l1Etl *L1ETL) handleBatch(batch *ETLBatch) error {
 	l1BlockHeaders := make([]database.L1BlockHeader, 0, len(batch.Headers))
 	for i := range batch.Headers {
 		if _, ok := batch.HeadersWithLog[batch.Headers[i].Hash()]; ok {
 			l1BlockHeaders = append(l1BlockHeaders, database.L1BlockHeader{BlockHeader: database.BlockHeaderFromHeader(&batch.Headers[i])})
 		}
 	}
+
 	if len(l1BlockHeaders) == 0 {
 		batch.Logger.Info("no l1 blocks with logs in batch")
-				continue
+		return nil
 	}
 
 	l1ContractEvents := make([]database.L1ContractEvent, len(batch.Logs))
 	for i := range batch.Logs {
 		timestamp := batch.HeaderMap[batch.Logs[i].BlockHash].Time
 		l1ContractEvents[i] = database.L1ContractEvent{ContractEvent: database.ContractEventFromLog(&batch.Logs[i], timestamp)}
+		l1Etl.ETL.metrics.RecordIndexedLog(batch.Logs[i].Address)
 	}
 
 	// Continually try to persist this batch. If it fails after 10 attempts, we simply error out
 	retryStrategy := &retry.ExponentialStrategy{Min: 1000, Max: 20_000, MaxJitter: 250}
-			if _, err := retry.Do[interface{}](ctx, 10, retryStrategy, func() (interface{}, error) {
+	if _, err := retry.Do[interface{}](l1Etl.resourceCtx, 10, retryStrategy, func() (interface{}, error) {
 		if err := l1Etl.db.Transaction(func(tx *database.DB) error {
 			if err := tx.Blocks.StoreL1BlockHeaders(l1BlockHeaders); err != nil {
 				return err
@@ -133,12 +182,11 @@ func (l1Etl *L1ETL) Start(ctx context.Context) error {
 			return nil
 		}); err != nil {
 			batch.Logger.Error("unable to persist batch", "err", err)
-					return nil, err
+			return nil, fmt.Errorf("unable to persist batch: %w", err)
 		}
 
 		l1Etl.ETL.metrics.RecordIndexedHeaders(len(l1BlockHeaders))
 		l1Etl.ETL.metrics.RecordIndexedLatestHeight(l1BlockHeaders[len(l1BlockHeaders)-1].Number)
-				l1Etl.ETL.metrics.RecordIndexedLogs(len(l1ContractEvents))
 
 		// a-ok!
 		return nil, nil
@@ -159,8 +207,7 @@ func (l1Etl *L1ETL) Start(ctx context.Context) error {
 		}
 	}
 	l1Etl.mu.Unlock()
-		}
-	}
+	return nil
 }
 
 // Notify returns a channel that'll receive a value every time new data has

indexer/etl/l1_etl_test.go

@@ -62,7 +62,7 @@ func TestL1ETLConstruction(t *testing.T) {
 			},
 			assertion: func(etl *L1ETL, err error) {
 				require.NoError(t, err)
-				require.Equal(t, etl.headerTraversal.LastHeader().ParentHash, common.HexToHash("0x69"))
+				require.Equal(t, etl.headerTraversal.LastTraversedHeader().ParentHash, common.HexToHash("0x69"))
 			},
 		},
 		{
@@ -94,7 +94,7 @@ func TestL1ETLConstruction(t *testing.T) {
 			},
 			assertion: func(etl *L1ETL, err error) {
 				require.NoError(t, err)
-				header := etl.headerTraversal.LastHeader()
+				header := etl.headerTraversal.LastTraversedHeader()
 
 				require.True(t, header.Number.Cmp(big.NewInt(69)) == 0)
 			},
@@ -108,7 +108,9 @@ func TestL1ETLConstruction(t *testing.T) {
 			logger := testlog.Logger(t, log.LvlInfo)
 			cfg := Config{StartHeight: ts.start}
 
-			etl, err := NewL1ETL(cfg, logger, ts.db.DB, etlMetrics, ts.client, ts.contracts)
+			etl, err := NewL1ETL(cfg, logger, ts.db.DB, etlMetrics, ts.client, ts.contracts, func(cause error) {
+				t.Fatalf("crit error: %v", cause)
+			})
 			test.assertion(etl, err)
 		})
 	}

indexer/etl/l2_etl.go

@@ -3,24 +3,34 @@ package etl
 import (
 	"context"
 	"errors"
+	"fmt"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+
 	"github.com/ethereum-optimism/optimism/indexer/config"
 	"github.com/ethereum-optimism/optimism/indexer/database"
 	"github.com/ethereum-optimism/optimism/indexer/node"
 	"github.com/ethereum-optimism/optimism/op-service/retry"
-	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/core/types"
-	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum-optimism/optimism/op-service/tasks"
 )
 
 type L2ETL struct {
 	ETL
 
+	// the batch handler may do work that we can interrupt on shutdown
+	resourceCtx    context.Context
+	resourceCancel context.CancelFunc
+
+	tasks tasks.Group
+
 	db *database.DB
 }
 
-func NewL2ETL(cfg Config, log log.Logger, db *database.DB, metrics Metricer, client node.EthClient, contracts config.L2Contracts) (*L2ETL, error) {
+func NewL2ETL(cfg Config, log log.Logger, db *database.DB, metrics Metricer, client node.EthClient,
+	contracts config.L2Contracts, shutdown context.CancelCauseFunc) (*L2ETL, error) {
 	log = log.New("etl", "l2")
 
 	zeroAddr := common.Address{}
@@ -54,7 +64,7 @@ func NewL2ETL(cfg Config, log log.Logger, db *database.DB, metrics Metricer, cli
 		log.Info("no indexed state, starting from genesis")
 	}
 
-	etlBatches := make(chan ETLBatch)
+	etlBatches := make(chan *ETLBatch)
 	etl := ETL{
 		loopInterval:     time.Duration(cfg.LoopIntervalMsec) * time.Millisecond,
 		headerBufferSize: uint64(cfg.HeaderBufferSize),
@@ -68,22 +78,57 @@ func NewL2ETL(cfg Config, log log.Logger, db *database.DB, metrics Metricer, cli
 		EthClient: client,
 	}
 
-	return &L2ETL{ETL: etl, db: db}, nil
+	resCtx, resCancel := context.WithCancel(context.Background())
+	return &L2ETL{
+		ETL:            etl,
+		resourceCtx:    resCtx,
+		resourceCancel: resCancel,
+		db:             db,
+		tasks: tasks.Group{HandleCrit: func(err error) {
+			shutdown(fmt.Errorf("critical error in L2 ETL: %w", err))
+		}},
+	}, nil
+}
+
+func (l2Etl *L2ETL) Close() error {
+	var result error
+	// close the producer
+	if err := l2Etl.ETL.Close(); err != nil {
+		result = errors.Join(result, fmt.Errorf("failed to close internal ETL: %w", err))
+	}
+	// tell the consumer it can stop what it's doing
+	l2Etl.resourceCancel()
+	// wait for consumer to pick up on closure of producer
+	if err := l2Etl.tasks.Wait(); err != nil {
+		result = errors.Join(result, fmt.Errorf("failed to await batch handler completion: %w", err))
+	}
+	return result
 }
 
-func (l2Etl *L2ETL) Start(ctx context.Context) error {
-	errCh := make(chan error, 1)
-	go func() {
-		errCh <- l2Etl.ETL.Start(ctx)
-	}()
+func (l2Etl *L2ETL) Start() error {
+	// start ETL batch producer
+	if err := l2Etl.ETL.Start(); err != nil {
+		return fmt.Errorf("failed to start internal ETL: %w", err)
+	}
 
+	// start ETL batch consumer
+	l2Etl.tasks.Go(func() error {
 		for {
-		select {
-		case err := <-errCh:
-			return err
+			// Index incoming batches (all L2 blocks)
+			batch, ok := <-l2Etl.etlBatches
+			if !ok {
+				l2Etl.log.Info("No more batches, shutting down L2 batch handler")
+				return nil
+			}
+			if err := l2Etl.handleBatch(batch); err != nil {
+				return fmt.Errorf("failed to handle batch, stopping L2 ETL: %w", err)
+			}
+		}
+	})
+	return nil
+}
 
-		// Index incoming batches (all L2 Blocks)
-		case batch := <-l2Etl.etlBatches:
+func (l2Etl *L2ETL) handleBatch(batch *ETLBatch) error {
 	l2BlockHeaders := make([]database.L2BlockHeader, len(batch.Headers))
 	for i := range batch.Headers {
 		l2BlockHeaders[i] = database.L2BlockHeader{BlockHeader: database.BlockHeaderFromHeader(&batch.Headers[i])}
@@ -93,11 +138,12 @@ func (l2Etl *L2ETL) Start(ctx context.Context) error {
 	for i := range batch.Logs {
 		timestamp := batch.HeaderMap[batch.Logs[i].BlockHash].Time
 		l2ContractEvents[i] = database.L2ContractEvent{ContractEvent: database.ContractEventFromLog(&batch.Logs[i], timestamp)}
+		l2Etl.ETL.metrics.RecordIndexedLog(batch.Logs[i].Address)
 	}
 
 	// Continually try to persist this batch. If it fails after 10 attempts, we simply error out
 	retryStrategy := &retry.ExponentialStrategy{Min: 1000, Max: 20_000, MaxJitter: 250}
-			if _, err := retry.Do[interface{}](ctx, 10, retryStrategy, func() (interface{}, error) {
+	if _, err := retry.Do[interface{}](l2Etl.resourceCtx, 10, retryStrategy, func() (interface{}, error) {
 		if err := l2Etl.db.Transaction(func(tx *database.DB) error {
 			if err := tx.Blocks.StoreL2BlockHeaders(l2BlockHeaders); err != nil {
 				return err
@@ -115,9 +161,6 @@ func (l2Etl *L2ETL) Start(ctx context.Context) error {
 
 		l2Etl.ETL.metrics.RecordIndexedHeaders(len(l2BlockHeaders))
 		l2Etl.ETL.metrics.RecordIndexedLatestHeight(l2BlockHeaders[len(l2BlockHeaders)-1].Number)
-				if len(l2ContractEvents) > 0 {
-					l2Etl.ETL.metrics.RecordIndexedLogs(len(l2ContractEvents))
-				}
 
 		// a-ok!
 		return nil, nil
@@ -126,6 +169,5 @@ func (l2Etl *L2ETL) Start(ctx context.Context) error {
 	}
 
 	batch.Logger.Info("indexed batch")
-		}
-	}
+	return nil
 }

indexer/indexer.toml

@@ -29,6 +29,7 @@ name = "$INDEXER_DB_NAME"
 [http]
 host = "127.0.0.1"
 port = 8080
+timeout = 10
 
 [metrics]
 host = "127.0.0.1"

indexer/node/metrics.go

@@ -12,7 +12,7 @@ import (
 )
 
 var (
-	MetricsNamespace = "rpc"
+	MetricsNamespace = "op_indexer_rpc"
 	batchMethod      = "<batch>"
 )
 

indexer/node/mocks.go

@@ -41,7 +41,10 @@ func (m *MockEthClient) StorageHash(address common.Address, blockNumber *big.Int
 	return args.Get(0).(common.Hash), args.Error(1)
 }
 
-func (m *MockEthClient) FilterLogs(query ethereum.FilterQuery) ([]types.Log, error) {
+func (m *MockEthClient) FilterLogs(query ethereum.FilterQuery) (Logs, error) {
 	args := m.Called(query)
-	return args.Get(0).([]types.Log), args.Error(1)
+	return args.Get(0).(Logs), args.Error(1)
+}
+
+func (m *MockEthClient) Close() {
 }

op-batcher/Dockerfile.dockerignore

+# ignore everything but the dockerfile, the op-stack-go base image performs the build
+*