Skip to content

N
nebula
Commit 1636250a authored by mergify[bot]'s avatar mergify[bot] Committed by GitHub
Browse files
Options

Merge branch 'develop' into sc/ctb-oo-fix-init

parents 01c947c0 19e581d8
Show whitespace changes
Inline Side-by-side
Showing with 1224 additions and 742 deletions
.changeset/lemon-files-rescue.md 0 → 100644
View file @ 1636250a
---
'@eth-optimism/indexer': minor
---
Bedrock support
.circleci/config.yml
View file @ 1636250a
......@@ -432,7 +432,7 @@ jobs:
name: Test
command: |
mkdir -p /test-results
gotestsum --junitfile /test-results/tests.xml
DB_USER=postgres gotestsum --junitfile /test-results/tests.xml
working_directory: <<parameters.working_directory>>
- when:
condition:
......
indexer/config.go
View file @ 1636250a
......@@ -4,6 +4,7 @@ import (
"errors"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/urfave/cli"
......@@ -20,15 +21,8 @@ var (
type Config struct {
/* Required Params */
// BuildEnv identifies the environment this binary is intended for, i.e.
// production, development, etc.
BuildEnv string
// EthNetworkName identifies the intended Ethereum network.
EthNetworkName string
// ChainID identifies the chain being indexed.
ChainID int64
ChainID uint64
// L1EthRpc is the HTTP provider URL for L1.
L1EthRpc string
......@@ -36,8 +30,8 @@ type Config struct {
// L2EthRpc is the HTTP provider URL for L1.
L2EthRpc string
// L2GenesisBlockHash is the l2 genesis block hash.
L2GenesisBlockHash string
// L1AddressManagerAddress is the address of the address manager for L1.
L1AddressManagerAddress string
// PollInterval is the delay between querying L2 for more transaction
// and creating a new batch.
......@@ -68,22 +62,8 @@ type Config struct {
// are printed using JSON.
LogTerminal bool
// SentryEnable if true, logs any error messages to sentry. SentryDsn
// must also be set if SentryEnable is true.
SentryEnable bool
// SentryDsn is the sentry Data Source Name.
SentryDsn string
// SentryTraceRate the frequency with which Sentry should flush buffered
// events.
SentryTraceRate time.Duration
// StartBlockNumber is the block number to start indexing from.
StartBlockNumber uint64
// StartBlockHash is the block hash to start indexing from.
StartBlockHash string
// L1StartBlockNumber is the block number to start indexing L1 from.
L1StartBlockNumber uint64
// ConfDepth is the number of confirmations after which headers are
// considered confirmed.
......@@ -111,6 +91,13 @@ type Config struct {
// DisableIndexer enables/disables the indexer.
DisableIndexer bool
// Bedrock enabled Bedrock indexing.
Bedrock bool
BedrockL1StandardBridgeAddress common.Address
BedrockOptimismPortalAddress common.Address
}
// NewConfig parses the Config from the provided flags or environment variables.
......@@ -118,26 +105,23 @@ type Config struct {
func NewConfig(ctx *cli.Context) (Config, error) {
cfg := Config{
/* Required Flags */
BuildEnv: ctx.GlobalString(flags.BuildEnvFlag.Name),
EthNetworkName: ctx.GlobalString(flags.EthNetworkNameFlag.Name),
ChainID: ctx.GlobalInt64(flags.ChainIDFlag.Name),
ChainID: ctx.GlobalUint64(flags.ChainIDFlag.Name),
L1EthRpc: ctx.GlobalString(flags.L1EthRPCFlag.Name),
L2EthRpc: ctx.GlobalString(flags.L2EthRPCFlag.Name),
L2GenesisBlockHash: ctx.GlobalString(flags.L2GenesisBlockHashFlag.Name),
L1AddressManagerAddress: ctx.GlobalString(flags.L1AddressManagerAddressFlag.Name),
DBHost: ctx.GlobalString(flags.DBHostFlag.Name),
DBPort: ctx.GlobalUint64(flags.DBPortFlag.Name),
DBUser: ctx.GlobalString(flags.DBUserFlag.Name),
DBPassword: ctx.GlobalString(flags.DBPasswordFlag.Name),
DBName: ctx.GlobalString(flags.DBNameFlag.Name),
/* Optional Flags */
Bedrock: ctx.GlobalBool(flags.BedrockFlag.Name),
BedrockL1StandardBridgeAddress: common.HexToAddress(ctx.GlobalString(flags.BedrockL1StandardBridgeAddress.Name)),
BedrockOptimismPortalAddress: common.HexToAddress(ctx.GlobalString(flags.BedrockOptimismPortalAddress.Name)),
DisableIndexer: ctx.GlobalBool(flags.DisableIndexer.Name),
LogLevel: ctx.GlobalString(flags.LogLevelFlag.Name),
LogTerminal: ctx.GlobalBool(flags.LogTerminalFlag.Name),
SentryEnable: ctx.GlobalBool(flags.SentryEnableFlag.Name),
SentryDsn: ctx.GlobalString(flags.SentryDsnFlag.Name),
SentryTraceRate: ctx.GlobalDuration(flags.SentryTraceRateFlag.Name),
StartBlockNumber: ctx.GlobalUint64(flags.StartBlockNumberFlag.Name),
StartBlockHash: ctx.GlobalString(flags.StartBlockHashFlag.Name),
L1StartBlockNumber: ctx.GlobalUint64(flags.L1StartBlockNumberFlag.Name),
ConfDepth: ctx.GlobalUint64(flags.ConfDepthFlag.Name),
MaxHeaderBatchSize: ctx.GlobalUint64(flags.MaxHeaderBatchSizeFlag.Name),
MetricsServerEnable: ctx.GlobalBool(flags.MetricsServerEnableFlag.Name),
......@@ -168,9 +152,8 @@ func ValidateConfig(cfg *Config) error {
return err
}
// Ensure the Sentry Data Source Name is set when using Sentry.
if cfg.SentryEnable && cfg.SentryDsn == "" {
return ErrSentryDSNNotSet
if cfg.Bedrock && (cfg.BedrockL1StandardBridgeAddress == common.Address{} || cfg.BedrockOptimismPortalAddress == common.Address{}) {
return errors.New("must specify l1 standard bridge and optimism portal addresses in bedrock mode")
}
return nil
......
indexer/db/db.go
View file @ 1636250a
......@@ -197,6 +197,12 @@ func (d *Database) AddIndexedL1Block(block *IndexedL1Block) error {
VALUES
($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
`
const updateWithdrawalStatement = `
UPDATE withdrawals SET (br_withdrawal_finalized_tx_hash, br_withdrawal_finalized_log_index, br_withdrawal_finalized_success) = ($1, $2, $3)
WHERE br_withdrawal_hash = $4
`
return txn(d.db, func(tx *sql.Tx) error {
_, err := tx.Exec(
insertBlockStatement,
......@@ -209,10 +215,7 @@ func (d *Database) AddIndexedL1Block(block *IndexedL1Block) error {
return err
}
if len(block.Deposits) == 0 {
return nil
}
if len(block.Deposits) > 0 {
for _, deposit := range block.Deposits {
_, err = tx.Exec(
insertDepositStatement,
......@@ -231,6 +234,22 @@ func (d *Database) AddIndexedL1Block(block *IndexedL1Block) error {
return err
}
}
}
if len(block.FinalizedWithdrawals) > 0 {
for _, wd := range block.FinalizedWithdrawals {
_, err = tx.Exec(
updateWithdrawalStatement,
wd.TxHash.String(),
wd.LogIndex,
wd.Success,
wd.WithdrawalHash.String(),
)
if err != nil {
return err
}
}
}
return nil
})
......@@ -459,19 +478,21 @@ func (d *Database) GetWithdrawalBatch(hash common.Hash) (*StateBatchJSON, error)
// GetWithdrawalsByAddress returns the list of Withdrawals indexed for the given
// address paginated by the given params.
func (d *Database) GetWithdrawalsByAddress(address common.Address, page PaginationParam) (*PaginatedWithdrawals, error) {
func (d *Database) GetWithdrawalsByAddress(address common.Address, page PaginationParam, state FinalizationState) (*PaginatedWithdrawals, error) {
selectWithdrawalsStatement := fmt.Sprintf(`
SELECT
withdrawals.guid, withdrawals.from_address, withdrawals.to_address,
withdrawals.amount, withdrawals.tx_hash, withdrawals.data,
withdrawals.l1_token, withdrawals.l2_token,
l2_tokens.name, l2_tokens.symbol, l2_tokens.decimals,
l2_blocks.number, l2_blocks.timestamp, withdrawals.br_withdrawal_hash
l2_blocks.number, l2_blocks.timestamp, withdrawals.br_withdrawal_hash,
withdrawals.br_withdrawal_finalized_tx_hash, withdrawals.br_withdrawal_finalized_log_index,
withdrawals.br_withdrawal_finalized_success
FROM withdrawals
INNER JOIN l2_blocks ON withdrawals.block_hash=l2_blocks.hash
INNER JOIN l2_tokens ON withdrawals.l2_token=l2_tokens.address
WHERE withdrawals.from_address = $1 %s ORDER BY l2_blocks.timestamp LIMIT $2 OFFSET $3;
`, FinalizationStateAny.SQL())
`, state.SQL())
var withdrawals []WithdrawalJSON
err := txn(d.db, func(tx *sql.Tx) error {
......@@ -485,13 +506,16 @@ func (d *Database) GetWithdrawalsByAddress(address common.Address, page Paginati
var withdrawal WithdrawalJSON
var l2Token Token
var wdHash sql.NullString
var finTxHash sql.NullString
var finLogIndex sql.NullInt32
var finSuccess sql.NullBool
if err := rows.Scan(
&withdrawal.GUID, &withdrawal.FromAddress, &withdrawal.ToAddress,
&withdrawal.Amount, &withdrawal.TxHash, &withdrawal.Data,
&withdrawal.L1Token, &l2Token.Address,
&l2Token.Name, &l2Token.Symbol, &l2Token.Decimals,
&withdrawal.BlockNumber, &withdrawal.BlockTimestamp,
&wdHash,
&wdHash, &finTxHash, &finLogIndex, &finSuccess,
); err != nil {
return err
}
......@@ -499,6 +523,16 @@ func (d *Database) GetWithdrawalsByAddress(address common.Address, page Paginati
if wdHash.Valid {
withdrawal.BedrockWithdrawalHash = &wdHash.String
}
if finTxHash.Valid {
withdrawal.BedrockFinalizedTxHash = &finTxHash.String
}
if finLogIndex.Valid {
idx := int(finLogIndex.Int32)
withdrawal.BedrockFinalizedLogIndex = &idx
}
if finSuccess.Valid {
withdrawal.BedrockFinalizedSuccess = &finSuccess.Bool
}
withdrawals = append(withdrawals, withdrawal)
}
......
indexer/db/eth.go
View file @ 1636250a
package db
import "github.com/ethereum/go-ethereum/common"
import (
"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
"github.com/ethereum/go-ethereum/common"
)
var ETHL1Address common.Address
......@@ -13,14 +16,10 @@ var ETHL1Token = &Token{
Decimals: 18,
}
// ETHL2Address is a placeholder address for differentiating ETH transactions
// from ERC20 transactions on L2.
var ETHL2Address = common.HexToAddress("0xDeadDeAddeAddEAddeadDEaDDEAdDeaDDeAD0000")
// ETHL2Token is a placeholder token for differentiating ETH transactions from
// ERC20 transactions on L2.
var ETHL2Token = &Token{
Address: ETHL2Address.String(),
Address: predeploys.LegacyERC20ETH,
Name: "Ethereum",
Symbol: "ETH",
Decimals: 18,
......
indexer/db/l1block.go
View file @ 1636250a
......@@ -11,6 +11,7 @@ type IndexedL1Block struct {
Number uint64
Timestamp uint64
Deposits []Deposit
FinalizedWithdrawals []FinalizedWithdrawal
}
// String returns the block hash for the indexed l1 block.
......
indexer/db/sql.go
View file @ 1636250a
......@@ -125,8 +125,8 @@ CREATE TABLE IF NOT EXISTS airdrops (
const updateWithdrawalsTable = `
ALTER TABLE withdrawals ADD COLUMN IF NOT EXISTS br_withdrawal_hash VARCHAR NULL;
ALTER TABLE withdrawals ADD COLUMN IF NOT EXISTS br_withdrawal_finalized_tx_hash VARCHAR NULL;
ALTER TABLE withdrawals ADD COLUMN IF NOT EXISTS br_withdrawal_finalized_log_index BOOLEAN NULL;
ALTER TABLE withdrawals ADD COLUMN IF NOT EXISTS br_withdrawal_success BOOLEAN NULL;
ALTER TABLE withdrawals ADD COLUMN IF NOT EXISTS br_withdrawal_finalized_log_index INTEGER NULL;
ALTER TABLE withdrawals ADD COLUMN IF NOT EXISTS br_withdrawal_finalized_success BOOLEAN NULL;
CREATE INDEX IF NOT EXISTS withdrawals_br_withdrawal_hash ON withdrawals(br_withdrawal_hash);
`
......
indexer/db/withdrawal.go
View file @ 1636250a
......@@ -40,6 +40,9 @@ type WithdrawalJSON struct {
TxHash string `json:"transactionHash"`
Batch *StateBatchJSON `json:"batch"`
BedrockWithdrawalHash *string `json:"bedrockWithdrawalHash"`
BedrockFinalizedTxHash *string `json:"bedrockFinalizedTxHash"`
BedrockFinalizedLogIndex *int `json:"bedrockFinalizedLogIndex"`
BedrockFinalizedSuccess *bool `json:"bedrockFinalizedSuccess"`
}
type FinalizationState int
......@@ -64,9 +67,9 @@ func ParseFinalizationState(in string) FinalizationState {
func (f FinalizationState) SQL() string {
switch f {
case FinalizationStateFinalized:
return "AND withdrawals.l1_block_hash IS NOT NULL"
return "AND withdrawals.br_withdrawal_finalized_tx_hash IS NOT NULL"
case FinalizationStateUnfinalized:
return "AND withdrawals.l2_block_hash IS NULL"
return "AND withdrawals.br_withdrawal_finalized_tx_hash IS NULL"
}
return ""
......
indexer/flags/flags.go
View file @ 1636250a
......@@ -46,11 +46,11 @@ var (
Required: true,
EnvVar: prefixEnvVar("L2_ETH_RPC"),
}
L2GenesisBlockHashFlag = cli.StringFlag{
Name: "l2-genesis-block-hash",
Usage: "Genesis block hash of the L2 chain",
L1AddressManagerAddressFlag = cli.StringFlag{
Name: "l1-address-manager-address",
Usage: "Address of the L1 address manager",
Required: true,
EnvVar: prefixEnvVar("L2_GENESIS_BLOCK_HASH"),
EnvVar: prefixEnvVar("L1_ADDRESS_MANAGER_ADDRESS"),
}
DBHostFlag = cli.StringFlag{
Name: "db-host",
......@@ -83,6 +83,23 @@ var (
EnvVar: prefixEnvVar("DB_NAME"),
}
/* Bedrock Flags */
BedrockFlag = cli.BoolFlag{
Name: "bedrock",
Usage: "Whether or not this indexer should operate in Bedrock mode",
EnvVar: prefixEnvVar("BEDROCK"),
}
BedrockL1StandardBridgeAddress = cli.BoolFlag{
Name: "bedrock.l1-standard-bridge-address",
Usage: "Address of the L1 standard bridge",
EnvVar: prefixEnvVar("BEDROCK_L1_STANDARD_BRIDGE"),
}
BedrockOptimismPortalAddress = cli.BoolFlag{
Name: "bedrock.portal-address",
Usage: "Address of the portal",
EnvVar: prefixEnvVar("BEDROCK_OPTIMISM_PORTAL"),
}
/* Optional Flags */
DisableIndexer = cli.BoolFlag{
......@@ -120,18 +137,12 @@ var (
Value: 50 * time.Millisecond,
EnvVar: prefixEnvVar("SENTRY_TRACE_RATE"),
}
StartBlockNumberFlag = cli.Uint64Flag{
L1StartBlockNumberFlag = cli.Uint64Flag{
Name: "start-block-number",
Usage: "The block number to start indexing from. Must be use together with start block hash",
Value: 0,
EnvVar: prefixEnvVar("START_BLOCK_NUMBER"),
}
StartBlockHashFlag = cli.StringFlag{
Name: "start-block-hash",
Usage: "The block hash to start indexing from. Must be use together with start block number",
Value: "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3",
EnvVar: prefixEnvVar("START_BLOCK_HASH"),
}
ConfDepthFlag = cli.Uint64Flag{
Name: "conf-depth",
Usage: "The number of confirmations after which headers are considered confirmed",
......@@ -181,7 +192,7 @@ var requiredFlags = []cli.Flag{
ChainIDFlag,
L1EthRPCFlag,
L2EthRPCFlag,
L2GenesisBlockHashFlag,
L1AddressManagerAddressFlag,
DBHostFlag,
DBPortFlag,
DBUserFlag,
......@@ -190,6 +201,9 @@ var requiredFlags = []cli.Flag{
}
var optionalFlags = []cli.Flag{
BedrockFlag,
BedrockL1StandardBridgeAddress,
BedrockOptimismPortalAddress,
DisableIndexer,
LogLevelFlag,
LogTerminalFlag,
......@@ -198,8 +212,7 @@ var optionalFlags = []cli.Flag{
SentryTraceRateFlag,
ConfDepthFlag,
MaxHeaderBatchSizeFlag,
StartBlockNumberFlag,
StartBlockHashFlag,
L1StartBlockNumberFlag,
RESTHostnameFlag,
RESTPortFlag,
MetricsServerEnableFlag,
......
indexer/indexer.go
View file @ 1636250a
......@@ -11,6 +11,7 @@ import (
"time"
"github.com/ethereum-optimism/optimism/indexer/services"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum-optimism/optimism/indexer/metrics"
"github.com/ethereum-optimism/optimism/indexer/server"
......@@ -22,7 +23,6 @@ import (
"github.com/ethereum/go-ethereum/ethclient"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/rpc"
sentry "github.com/getsentry/sentry-go"
"github.com/gorilla/mux"
"github.com/urfave/cli"
)
......@@ -44,15 +44,9 @@ func Main(gitVersion string) func(ctx *cli.Context) error {
return err
}
// The call to defer is done here so that any errors logged from
// this point on are posted to Sentry before exiting.
if cfg.SentryEnable {
defer sentry.Flush(2 * time.Second)
}
log.Info("Initializing indexer")
indexer, err := NewIndexer(cfg, gitVersion)
indexer, err := NewIndexer(cfg)
if err != nil {
log.Error("Unable to create indexer", "error", err)
return err
......@@ -87,32 +81,18 @@ type Indexer struct {
router *mux.Router
metrics *metrics.Metrics
db *database.Database
server *http.Server
}
// NewIndexer initializes the Indexer, gathering any resources
// that will be needed by the TxIndexer and StateIndexer
// sub-services.
func NewIndexer(cfg Config, gitVersion string) (*Indexer, error) {
func NewIndexer(cfg Config) (*Indexer, error) {
ctx := context.Background()
// Set up our logging. If Sentry is enabled, we will use our custom
// log handler that logs to stdout and forwards any error messages to
// Sentry for collection. Otherwise, logs will only be posted to stdout.
var logHandler log.Handler
if cfg.SentryEnable {
err := sentry.Init(sentry.ClientOptions{
Dsn: cfg.SentryDsn,
Environment: cfg.EthNetworkName,
Release: "indexer@" + gitVersion,
TracesSampleRate: traceRateToFloat64(cfg.SentryTraceRate),
Debug: false,
})
if err != nil {
return nil, err
}
logHandler = SentryStreamHandler(os.Stdout, log.JSONFormat())
} else if cfg.LogTerminal {
if cfg.LogTerminal {
logHandler = log.StreamHandler(os.Stdout, log.TerminalFormat(true))
} else {
logHandler = log.StreamHandler(os.Stdout, log.JSONFormat())
......@@ -149,8 +129,11 @@ func NewIndexer(cfg Config, gitVersion string) (*Indexer, error) {
log.Info("metrics server enabled", "host", cfg.MetricsHostname, "port", cfg.MetricsPort)
}
dsn := fmt.Sprintf("host=%s port=%d user=%s dbname=%s sslmode=disable",
cfg.DBHost, cfg.DBPort, cfg.DBUser, cfg.DBName)
dsn := fmt.Sprintf("host=%s port=%d dbname=%s sslmode=disable",
cfg.DBHost, cfg.DBPort, cfg.DBName)
if cfg.DBUser != "" {
dsn += fmt.Sprintf(" user=%s", cfg.DBUser)
}
if cfg.DBPassword != "" {
dsn += fmt.Sprintf(" password=%s", cfg.DBPassword)
}
......@@ -159,17 +142,32 @@ func NewIndexer(cfg Config, gitVersion string) (*Indexer, error) {
return nil, err
}
var addrManager services.AddressManager
if cfg.Bedrock {
addrManager, err = services.NewBedrockAddresses(
l1Client,
cfg.BedrockL1StandardBridgeAddress,
cfg.BedrockOptimismPortalAddress,
)
} else {
addrManager, err = services.NewLegacyAddresses(l1Client, common.HexToAddress(cfg.L1AddressManagerAddress))
}
if err != nil {
return nil, err
}
l1IndexingService, err := l1.NewService(l1.ServiceConfig{
Context: ctx,
Metrics: m,
L1Client: l1Client,
RawL1Client: rawl1Client,
ChainID: big.NewInt(cfg.ChainID),
ChainID: new(big.Int).SetUint64(cfg.ChainID),
AddressManager: addrManager,
DB: db,
ConfDepth: cfg.ConfDepth,
MaxHeaderBatchSize: cfg.MaxHeaderBatchSize,
StartBlockNumber: cfg.StartBlockNumber,
StartBlockHash: cfg.StartBlockHash,
StartBlockNumber: cfg.L1StartBlockNumber,
Bedrock: cfg.Bedrock,
})
if err != nil {
return nil, err
......@@ -184,7 +182,7 @@ func NewIndexer(cfg Config, gitVersion string) (*Indexer, error) {
ConfDepth: cfg.ConfDepth,
MaxHeaderBatchSize: cfg.MaxHeaderBatchSize,
StartBlockNumber: uint64(0),
StartBlockHash: cfg.L2GenesisBlockHash,
Bedrock: cfg.Bedrock,
})
if err != nil {
return nil, err
......@@ -200,6 +198,7 @@ func NewIndexer(cfg Config, gitVersion string) (*Indexer, error) {
airdropService: services.NewAirdrop(db, m),
router: mux.NewRouter(),
metrics: m,
db: db,
}, nil
}
......@@ -212,7 +211,7 @@ func (b *Indexer) Serve() error {
b.router.HandleFunc("/v1/l1/status", b.l1IndexingService.GetIndexerStatus).Methods("GET")
b.router.HandleFunc("/v1/l2/status", b.l2IndexingService.GetIndexerStatus).Methods("GET")
b.router.HandleFunc("/v1/deposits/0x{address:[a-fA-F0-9]{40}}", b.l1IndexingService.GetDeposits).Methods("GET")
b.router.HandleFunc("/v1/withdrawal/0x{hash:[a-fA-F0-9]{64}}", b.l2IndexingService.GetWithdrawalStatus).Methods("GET")
b.router.HandleFunc("/v1/withdrawal/0x{hash:[a-fA-F0-9]{64}}", b.l2IndexingService.GetWithdrawalBatch).Methods("GET")
b.router.HandleFunc("/v1/withdrawals/0x{address:[a-fA-F0-9]{40}}", b.l2IndexingService.GetWithdrawals).Methods("GET")
b.router.HandleFunc("/v1/airdrops/0x{address:[a-fA-F0-9]{40}}", b.airdropService.GetAirdrop)
b.router.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
......@@ -228,8 +227,27 @@ func (b *Indexer) Serve() error {
port := strconv.FormatUint(b.cfg.RESTPort, 10)
addr := net.JoinHostPort(b.cfg.RESTHostname, port)
b.server = &http.Server{
Addr: addr,
Handler: middleware(c.Handler(b.router)),
}
errCh := make(chan error, 1)
go func() {
errCh <- b.server.ListenAndServe()
}()
// Capture server startup errors
<-time.After(10 * time.Millisecond)
select {
case err := <-errCh:
return err
default:
log.Info("indexer REST server listening on", "addr", addr)
return http.ListenAndServe(addr, middleware(c.Handler(b.router)))
return nil
}
}
// Start starts the starts the indexing service on L1 and L2 chains and also
......@@ -253,6 +271,14 @@ func (b *Indexer) Start() error {
// Stop stops the indexing service on L1 and L2 chains.
func (b *Indexer) Stop() {
b.db.Close()
if b.server != nil {
// background context here so it waits for
// conns to close
_ = b.server.Shutdown(context.Background())
}
if !b.cfg.DisableIndexer {
b.l1IndexingService.Stop()
b.l2IndexingService.Stop()
......@@ -274,14 +300,3 @@ func dialEthClientWithTimeout(ctx context.Context, url string) (
}
return ethclient.NewClient(c), c, nil
}
// traceRateToFloat64 converts a time.Duration into a valid float64 for the
// Sentry client. The client only accepts values between 0.0 and 1.0, so this
// method clamps anything greater than 1 second to 1.0.
func traceRateToFloat64(rate time.Duration) float64 {
rate64 := float64(rate) / float64(time.Second)
if rate64 > 1.0 {
rate64 = 1.0
}
return rate64
}
indexer/integration_tests/bedrock_test.go 0 → 100644
View file @ 1636250a
package integration_tests
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"math/big"
"net/http"
"os"
"testing"
"time"
"github.com/ethereum-optimism/optimism/indexer"
"github.com/ethereum-optimism/optimism/indexer/db"
"github.com/ethereum-optimism/optimism/indexer/services/l1"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
op_e2e "github.com/ethereum-optimism/optimism/op-e2e"
"github.com/ethereum-optimism/optimism/op-e2e/e2eutils"
"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
"github.com/ethereum-optimism/optimism/op-node/withdrawals"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rpc"
"github.com/stretchr/testify/require"
_ "github.com/lib/pq"
)
func TestBedrockIndexer(t *testing.T) {
dbParams := createTestDB(t)
cfg := op_e2e.DefaultSystemConfig(t)
cfg.DeployConfig.FinalizationPeriodSeconds = 2
sys, err := cfg.Start()
require.NoError(t, err)
defer sys.Close()
l1Client := sys.Clients["l1"]
l2Client := sys.Clients["sequencer"]
fromAddr := cfg.Secrets.Addresses().Alice
// wait a couple of blocks
require.NoError(t, e2eutils.WaitBlock(e2eutils.TimeoutCtx(t, 30*time.Second), l2Client, 10))
l1SB, err := bindings.NewL1StandardBridge(predeploys.DevL1StandardBridgeAddr, l1Client)
require.NoError(t, err)
l2SB, err := bindings.NewL2StandardBridge(predeploys.L2StandardBridgeAddr, l2Client)
require.NoError(t, err)
portal, err := bindings.NewOptimismPortal(predeploys.DevOptimismPortalAddr, l1Client)
require.NoError(t, err)
l1Opts, err := bind.NewKeyedTransactorWithChainID(cfg.Secrets.Alice, cfg.L1ChainIDBig())
require.NoError(t, err)
l2Opts, err := bind.NewKeyedTransactorWithChainID(cfg.Secrets.Alice, cfg.L2ChainIDBig())
require.NoError(t, err)
idxrCfg := indexer.Config{
ChainID: cfg.DeployConfig.L1ChainID,
L1EthRpc: sys.Nodes["l1"].HTTPEndpoint(),
L2EthRpc: sys.Nodes["sequencer"].HTTPEndpoint(),
PollInterval: time.Second,
DBHost: dbParams.Host,
DBPort: dbParams.Port,
DBUser: dbParams.User,
DBPassword: dbParams.Password,
DBName: dbParams.Name,
LogLevel: "info",
LogTerminal: true,
L1StartBlockNumber: 0,
ConfDepth: 1,
MaxHeaderBatchSize: 2,
RESTHostname: "127.0.0.1",
RESTPort: 7980,
DisableIndexer: false,
Bedrock: true,
BedrockL1StandardBridgeAddress: predeploys.DevL1StandardBridgeAddr,
BedrockOptimismPortalAddress: predeploys.DevOptimismPortalAddr,
}
idxr, err := indexer.NewIndexer(idxrCfg)
require.NoError(t, err)
errCh := make(chan error, 1)
go func() {
errCh <- idxr.Start()
}()
t.Cleanup(func() {
idxr.Stop()
require.NoError(t, <-errCh)
})
makeURL := func(path string) string {
return fmt.Sprintf("http://%s:%d/%s", idxrCfg.RESTHostname, idxrCfg.RESTPort, path)
}
t.Run("deposit ETH", func(t *testing.T) {
l1Opts.Value = big.NewInt(params.Ether)
depTx, err := l1SB.DepositETH(l1Opts, 200_000, nil)
require.NoError(t, err)
depReceipt, err := e2eutils.WaitReceiptOK(e2eutils.TimeoutCtx(t, 10*time.Second), l1Client, depTx.Hash())
require.NoError(t, err)
require.Greaterf(t, len(depReceipt.Logs), 0, "must have logs")
var l2Hash common.Hash
for _, eLog := range depReceipt.Logs {
if len(eLog.Topics) == 0 || eLog.Topics[0] != derive.DepositEventABIHash {
continue
}
depLog, err := derive.UnmarshalDepositLogEvent(eLog)
require.NoError(t, err)
tx := types.NewTx(depLog)
l2Hash = tx.Hash()
}
require.NotEqual(t, common.Hash{}, l2Hash)
_, err = e2eutils.WaitReceiptOK(e2eutils.TimeoutCtx(t, 15*time.Second), l2Client, l2Hash)
require.NoError(t, err)
// Poll for indexer deposit
var depPage *db.PaginatedDeposits
require.NoError(t, e2eutils.WaitFor(e2eutils.TimeoutCtx(t, 30*time.Second), 100*time.Millisecond, func() (bool, error) {
res := new(db.PaginatedDeposits)
err := getJSON(makeURL(fmt.Sprintf("v1/deposits/%s", fromAddr)), res)
if err != nil {
return false, err
}
if len(res.Deposits) == 0 {
return false, nil
}
depPage = res
return true, nil
}))
// Make sure deposit is what we expect
require.Equal(t, 1, len(depPage.Deposits))
deposit := depPage.Deposits[0]
require.Equal(t, big.NewInt(params.Ether).String(), deposit.Amount)
require.Equal(t, depTx.Hash().String(), deposit.TxHash)
require.Equal(t, depReceipt.BlockNumber.Uint64(), deposit.BlockNumber)
require.Equal(t, fromAddr.String(), deposit.FromAddress)
require.Equal(t, fromAddr.String(), deposit.ToAddress)
require.EqualValues(t, db.ETHL1Token, deposit.L1Token)
require.Equal(t, l1.ZeroAddress.String(), deposit.L2Token)
require.NotEmpty(t, deposit.GUID)
// Perform withdrawal through bridge
l2Opts.Value = big.NewInt(0.5 * params.Ether)
wdTx, err := l2SB.Withdraw(l2Opts, predeploys.LegacyERC20ETHAddr, big.NewInt(0.5*params.Ether), 0, nil)
require.NoError(t, err)
wdReceipt, err := e2eutils.WaitReceiptOK(e2eutils.TimeoutCtx(t, 30*time.Second), l2Client, wdTx.Hash())
require.NoError(t, err)
var wdPage *db.PaginatedWithdrawals
require.NoError(t, e2eutils.WaitFor(e2eutils.TimeoutCtx(t, 30*time.Second), 100*time.Millisecond, func() (bool, error) {
res := new(db.PaginatedWithdrawals)
err := getJSON(makeURL(fmt.Sprintf("v1/withdrawals/%s", fromAddr)), res)
if err != nil {
return false, err
}
if len(res.Withdrawals) == 0 {
return false, nil
}
wdPage = res
return true, nil
}))
require.Equal(t, 1, len(wdPage.Withdrawals))
withdrawal := wdPage.Withdrawals[0]
require.Nil(t, withdrawal.BedrockFinalizedTxHash)
require.Equal(t, big.NewInt(0.5*params.Ether).String(), withdrawal.Amount)
require.Equal(t, wdTx.Hash().String(), withdrawal.TxHash)
require.Equal(t, wdReceipt.BlockNumber.Uint64(), withdrawal.BlockNumber)
// use fromaddr twice here because the user is withdrawing
// to themselves
require.Equal(t, fromAddr.String(), withdrawal.FromAddress)
require.Equal(t, fromAddr.String(), withdrawal.ToAddress)
require.EqualValues(t, l1.ZeroAddress.String(), withdrawal.L1Token)
require.Equal(t, db.ETHL2Token, withdrawal.L2Token)
require.NotEmpty(t, withdrawal.GUID)
finBlockNum, err := withdrawals.WaitForFinalizationPeriod(
e2eutils.TimeoutCtx(t, time.Minute),
l1Client,
predeploys.DevOptimismPortalAddr,
wdReceipt.BlockNumber,
)
require.NoError(t, err)
finHeader, err := l2Client.HeaderByNumber(context.Background(), big.NewInt(int64(finBlockNum)))
require.NoError(t, err)
rpcClient, err := rpc.Dial(sys.Nodes["sequencer"].HTTPEndpoint())
require.NoError(t, err)
proofClient := withdrawals.NewClient(rpcClient)
wParams, err := withdrawals.FinalizeWithdrawalParameters(context.Background(), proofClient, wdTx.Hash(), finHeader)
require.NoError(t, err)
l1Opts.Value = big.NewInt(0)
finTx, err := portal.FinalizeWithdrawalTransaction(
l1Opts,
bindings.TypesWithdrawalTransaction{
Nonce: wParams.Nonce,
Sender: wParams.Sender,
Target: wParams.Target,
Value: wParams.Value,
GasLimit: wParams.GasLimit,
Data: wParams.Data,
},
wParams.BlockNumber,
wParams.OutputRootProof,
wParams.WithdrawalProof,
)
require.NoError(t, err)
finReceipt, err := e2eutils.WaitReceiptOK(e2eutils.TimeoutCtx(t, time.Minute), l1Client, finTx.Hash())
require.NoError(t, err)
wdPage = nil
require.NoError(t, e2eutils.WaitFor(e2eutils.TimeoutCtx(t, 30*time.Second), 100*time.Millisecond, func() (bool, error) {
res := new(db.PaginatedWithdrawals)
err := getJSON(makeURL(fmt.Sprintf("v1/withdrawals/%s", fromAddr)), res)
if err != nil {
return false, err
}
if res.Withdrawals[0].BedrockFinalizedTxHash == nil {
return false, nil
}
wdPage = res
return true, nil
}))
wd := wdPage.Withdrawals[0]
require.Equal(t, finReceipt.TxHash.String(), *wd.BedrockFinalizedTxHash)
require.True(t, *wd.BedrockFinalizedSuccess)
wdPage = new(db.PaginatedWithdrawals)
err = getJSON(makeURL(fmt.Sprintf("v1/withdrawals/%s?finalized=false", fromAddr)), wdPage)
require.NoError(t, err)
require.Equal(t, 0, len(wdPage.Withdrawals))
})
}
type testDBParams struct {
Host string
Port uint64
User string
Password string
Name string
}
func createTestDB(t *testing.T) *testDBParams {
user := os.Getenv("DB_USER")
name := fmt.Sprintf("indexer_test_%d", time.Now().Unix())
dsn := "postgres://"
if user != "" {
dsn += user
dsn += "@"
}
dsn += "localhost:5432?sslmode=disable"
pg, err := sql.Open(
"postgres",
dsn,
)
require.NoError(t, err)
_, err = pg.Exec("CREATE DATABASE " + name)
require.NoError(t, err)
t.Cleanup(func() {
_, err = pg.Exec("DROP DATABASE " + name)
require.NoError(t, err)
pg.Close()
})
return &testDBParams{
Host: "localhost",
Port: 5432,
Name: name,
User: user,
}
}
func getJSON(url string, out interface{}) error {
res, err := http.Get(url)
if err != nil {
return err
}
if res.StatusCode != 200 {
return fmt.Errorf("non-200 status code %d", res.StatusCode)
}
defer res.Body.Close()
dec := json.NewDecoder(res.Body)
return dec.Decode(out)
}
indexer/metrics/metrics.go
View file @ 1636250a
......@@ -21,6 +21,8 @@ type Metrics struct {
WithdrawalsCount *prometheus.CounterVec
StateBatchesCount prometheus.Counter
L1CatchingUp prometheus.Gauge
L2CatchingUp prometheus.Gauge
......@@ -72,6 +74,12 @@ func NewMetrics(monitoredTokens map[string]string) *Metrics {
"symbol",
}),
StateBatchesCount: promauto.NewCounter(prometheus.CounterOpts{
Name: "state_batches_count",
Help: "The number of state batches indexed.",
Namespace: metricsNamespace,
}),
L1CatchingUp: promauto.NewGauge(prometheus.GaugeOpts{
Name: "l1_catching_up",
Help: "Whether or not L1 is far behind the chain tip.",
......@@ -160,6 +168,10 @@ func (m *Metrics) RecordWithdrawal(addr common.Address) {
m.WithdrawalsCount.WithLabelValues(sym).Inc()
}
func (m *Metrics) RecordStateBatches(count int) {
m.StateBatchesCount.Add(float64(count))
}
func (m *Metrics) SetL1CatchingUp(state bool) {
var catchingUp float64
if state {
......
indexer/sentry_log.go deleted 100644 → 0
View file @ 01c947c0
package indexer
import (
"errors"
"io"
"github.com/ethereum/go-ethereum/log"
"github.com/getsentry/sentry-go"
)
var jsonFmt = log.JSONFormat()
// SentryStreamHandler creates a log.Handler that behaves similarly to
// log.StreamHandler, however it writes any log with severity greater than or
// equal to log.LvlError to Sentry. In that case, the passed log.Record is
// encoded using JSON rather than the default terminal output, so that it can be
// captured for debugging in the Sentry dashboard.
func SentryStreamHandler(wr io.Writer, fmtr log.Format) log.Handler {
h := log.FuncHandler(func(r *log.Record) error {
_, err := wr.Write(fmtr.Format(r))
// If this record's severity is log.LvlError or higher,
// serialize the record using JSON and write it to Sentry. We
// also capture the error message separately so that it's easy
// to parse what the error is in the dashboard.
//
// NOTE: The log.Lvl* constants are defined in reverse order of
// their severity, i.e. zero (log.LvlCrit) is the highest
// severity.
if r.Lvl <= log.LvlError {
sentry.WithScope(func(scope *sentry.Scope) {
scope.SetExtra("context", jsonFmt.Format(r))
sentry.CaptureException(errors.New(r.Msg))
})
}
return err
})
return log.LazyHandler(log.SyncHandler(h))
}
indexer/services/addresses.go 0 → 100644
View file @ 1636250a
package services
import (
"github.com/ethereum-optimism/optimism/indexer/bindings/legacy/scc"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
"github.com/ethereum/go-ethereum/common"
)
type AddressManager interface {
L1StandardBridge() (common.Address, *bindings.L1StandardBridge)
StateCommitmentChain() (common.Address, *scc.StateCommitmentChain)
OptimismPortal() (common.Address, *bindings.OptimismPortal)
}
type LegacyAddresses struct {
l1SB *bindings.L1StandardBridge
l1SBAddr common.Address
scc *scc.StateCommitmentChain
sccAddr common.Address
}
var _ AddressManager = (*LegacyAddresses)(nil)
func NewLegacyAddresses(client bind.ContractBackend, addrMgrAddr common.Address) (AddressManager, error) {
mgr, err := bindings.NewAddressManager(addrMgrAddr, client)
if err != nil {
return nil, err
}
l1SBAddr, err := mgr.GetAddress(nil, "Proxy__OVM_L1StandardBridge")
if err != nil {
return nil, err
}
sccAddr, err := mgr.GetAddress(nil, "StateCommitmentChain")
if err != nil {
return nil, err
}
l1SB, err := bindings.NewL1StandardBridge(l1SBAddr, client)
if err != nil {
return nil, err
}
sccContract, err := scc.NewStateCommitmentChain(sccAddr, client)
if err != nil {
return nil, err
}
return &LegacyAddresses{
l1SB: l1SB,
l1SBAddr: l1SBAddr,
scc: sccContract,
sccAddr: sccAddr,
}, nil
}
func (a *LegacyAddresses) L1StandardBridge() (common.Address, *bindings.L1StandardBridge) {
return a.l1SBAddr, a.l1SB
}
func (a *LegacyAddresses) StateCommitmentChain() (common.Address, *scc.StateCommitmentChain) {
return a.sccAddr, a.scc
}
func (a *LegacyAddresses) OptimismPortal() (common.Address, *bindings.OptimismPortal) {
panic("OptimismPortal not configured on legacy networks - this is a programmer error")
}
type BedrockAddresses struct {
l1SB *bindings.L1StandardBridge
l1SBAddr common.Address
portal *bindings.OptimismPortal
portalAddr common.Address
}
var _ AddressManager = (*BedrockAddresses)(nil)
func NewBedrockAddresses(client bind.ContractBackend, l1SBAddr, portalAddr common.Address) (AddressManager, error) {
l1SB, err := bindings.NewL1StandardBridge(l1SBAddr, client)
if err != nil {
return nil, err
}
portal, err := bindings.NewOptimismPortal(portalAddr, client)
if err != nil {
return nil, err
}
return &BedrockAddresses{
l1SB: l1SB,
l1SBAddr: l1SBAddr,
portal: portal,
portalAddr: portalAddr,
}, nil
}
func (b *BedrockAddresses) L1StandardBridge() (common.Address, *bindings.L1StandardBridge) {
return b.l1SBAddr, b.l1SB
}
func (b *BedrockAddresses) StateCommitmentChain() (common.Address, *scc.StateCommitmentChain) {
panic("SCC not configured on legacy networks - this is a programmer error")
}
func (b *BedrockAddresses) OptimismPortal() (common.Address, *bindings.OptimismPortal) {
return b.portalAddr, b.portal
}
indexer/services/l1/bridge/bridge.go
View file @ 1636250a
......@@ -5,93 +5,85 @@ import (
"errors"
"math/big"
"github.com/ethereum-optimism/optimism/indexer/bindings/legacy/scc"
"github.com/ethereum-optimism/optimism/indexer/db"
"github.com/ethereum-optimism/optimism/indexer/services"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
"github.com/ethereum/go-ethereum/common"
)
// DepositsMap is a collection of deposit objects keyed
// on block hashes.
type DepositsMap map[common.Hash][]db.Deposit
type WithdrawalsMap map[common.Hash][]db.Withdrawal // Finalizations
// WithdrawalsMap is a collection of withdrawal objects keyed
// on block hashes.
type InitiatedWithdrawalMap map[common.Hash][]db.Withdrawal
// FinalizedWithdrawalsMap is a collection of finalized withdrawal
// objected keyed on block hashes.
type FinalizedWithdrawalsMap map[common.Hash][]db.FinalizedWithdrawal
type Bridge interface {
Address() common.Address
GetDepositsByBlockRange(uint64, uint64) (DepositsMap, error)
GetWithdrawalsByBlockRange(uint64, uint64) (WithdrawalsMap, error)
GetDepositsByBlockRange(context.Context, uint64, uint64) (DepositsMap, error)
String() string
}
type implConfig struct {
name string
impl string
addr string
}
var defaultBridgeCfgs = map[uint64][]*implConfig{
// Devnet
900: {
{"Standard", "StandardBridge", predeploys.DevL1StandardBridge},
{"ETH", "ETHBridge", predeploys.DevL1StandardBridge},
},
// Goerli
5: {
{"Standard", "StandardBridge", "0xFf94B6C486350aD92561Ba09bad3a59df764Da92"},
{"ETH", "ETHBridge", "0xFf94B6C486350aD92561Ba09bad3a59df764Da92"},
},
addr common.Address
}
var customBridgeCfgs = map[uint64][]*implConfig{
// Mainnet
1: {
{"BitBTC", "StandardBridge", "0xaBA2c5F108F7E820C049D5Af70B16ac266c8f128"},
{"DAI", "StandardBridge", "0x10E6593CDda8c58a1d0f14C5164B376352a55f2F"},
{"BitBTC", "StandardBridge", common.HexToAddress("0xaBA2c5F108F7E820C049D5Af70B16ac266c8f128")},
{"DAI", "StandardBridge", common.HexToAddress("0x10E6593CDda8c58a1d0f14C5164B376352a55f2F")},
{"wstETH", "StandardBridge", common.HexToAddress("0x76943C0D61395d8F2edF9060e1533529cAe05dE6")},
},
// Kovan
42: {
{"BitBTC", "StandardBridge", "0x0b651A42F32069d62d5ECf4f2a7e5Bd3E9438746"},
{"USX", "StandardBridge", "0x40E862341b2416345F02c41Ac70df08525150dC7"},
{"DAI", "StandardBridge", "0xb415e822C4983ecD6B1c1596e8a5f976cf6CD9e3"},
{"BitBTC", "StandardBridge", common.HexToAddress("0x0b651A42F32069d62d5ECf4f2a7e5Bd3E9438746")},
{"USX", "StandardBridge", common.HexToAddress("0x40E862341b2416345F02c41Ac70df08525150dC7")},
{"DAI", "StandardBridge", common.HexToAddress("0xb415e822C4983ecD6B1c1596e8a5f976cf6CD9e3")},
{"wstETH", "StandardBridge", common.HexToAddress("0x65321bf24210b81500230dCEce14Faa70a9f50a7")},
},
}
func BridgesByChainID(chainID *big.Int, client bind.ContractBackend, ctx context.Context) (map[string]Bridge, error) {
allCfgs := make([]*implConfig, 0)
allCfgs = append(allCfgs, defaultBridgeCfgs[chainID.Uint64()]...)
func BridgesByChainID(chainID *big.Int, client bind.ContractBackend, addrs services.AddressManager) (map[string]Bridge, error) {
l1SBAddr, _ := addrs.L1StandardBridge()
allCfgs := []*implConfig{
{"Standard", "StandardBridge", l1SBAddr},
{"ETH", "ETHBridge", l1SBAddr},
}
allCfgs = append(allCfgs, customBridgeCfgs[chainID.Uint64()]...)
bridges := make(map[string]Bridge)
for _, bridge := range allCfgs {
switch bridge.impl {
case "StandardBridge":
l1StandardBridgeAddress := common.HexToAddress(bridge.addr)
l1StandardBridgeFilter, err := bindings.NewL1StandardBridgeFilterer(l1StandardBridgeAddress, client)
l1SB, err := bindings.NewL1StandardBridge(bridge.addr, client)
if err != nil {
return nil, err
}
standardBridge := &StandardBridge{
name: bridge.name,
ctx: ctx,
address: l1StandardBridgeAddress,
client: client,
filterer: l1StandardBridgeFilter,
address: bridge.addr,
contract: l1SB,
}
bridges[bridge.name] = standardBridge
case "ETHBridge":
l1StandardBridgeAddress := common.HexToAddress(bridge.addr)
l1EthBridgeFilter, err := bindings.NewL1StandardBridgeFilterer(l1StandardBridgeAddress, client)
l1SB, err := bindings.NewL1StandardBridge(bridge.addr, client)
if err != nil {
return nil, err
}
ethBridge := &EthBridge{
name: bridge.name,
ctx: ctx,
address: l1StandardBridgeAddress,
client: client,
filterer: l1EthBridgeFilter,
address: bridge.addr,
contract: l1SB,
}
bridges[bridge.name] = ethBridge
default:
......@@ -100,3 +92,12 @@ func BridgesByChainID(chainID *big.Int, client bind.ContractBackend, ctx context
}
return bridges, nil
}
func StateCommitmentChainScanner(client bind.ContractFilterer, addrs services.AddressManager) (*scc.StateCommitmentChainFilterer, error) {
sccAddr, _ := addrs.StateCommitmentChain()
filter, err := scc.NewStateCommitmentChainFilterer(sccAddr, client)
if err != nil {
return nil, err
}
return filter, nil
}
indexer/services/l1/bridge/bridge.go.orig 0 → 100644
View file @ 1636250a
package bridge
import (
"context"
"errors"
"math/big"
"github.com/ethereum-optimism/optimism/indexer/bindings/legacy/scc"
"github.com/ethereum-optimism/optimism/indexer/db"
"github.com/ethereum-optimism/optimism/indexer/services"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
"github.com/ethereum/go-ethereum/common"
)
// DepositsMap is a collection of deposit objects keyed
// on block hashes.
type DepositsMap map[common.Hash][]db.Deposit
<<<<<<< HEAD
// WithdrawalsMap is a collection of withdrawal objects keyed
// on block hashes.
type WithdrawalsMap map[common.Hash][]db.Withdrawal
// FinalizedWithdrawalsMap is a collection of finalized withdrawal
// objected keyed on block hashes.
=======
type InitiatedWithdrawalMap map[common.Hash][]db.Withdrawal
>>>>>>> 22c039efc (indexer: Upgrade L1 services)
type FinalizedWithdrawalsMap map[common.Hash][]db.FinalizedWithdrawal
type Bridge interface {
Address() common.Address
GetDepositsByBlockRange(context.Context, uint64, uint64) (DepositsMap, error)
String() string
}
type implConfig struct {
name string
impl string
addr common.Address
}
var customBridgeCfgs = map[uint64][]*implConfig{
// Mainnet
1: {
{"BitBTC", "StandardBridge", common.HexToAddress("0xaBA2c5F108F7E820C049D5Af70B16ac266c8f128")},
{"DAI", "StandardBridge", common.HexToAddress("0x10E6593CDda8c58a1d0f14C5164B376352a55f2F")},
{"wstETH", "StandardBridge", common.HexToAddress("0x76943C0D61395d8F2edF9060e1533529cAe05dE6")},
},
// Kovan
42: {
{"BitBTC", "StandardBridge", common.HexToAddress("0x0b651A42F32069d62d5ECf4f2a7e5Bd3E9438746")},
{"USX", "StandardBridge", common.HexToAddress("0x40E862341b2416345F02c41Ac70df08525150dC7")},
{"DAI", "StandardBridge", common.HexToAddress("0xb415e822C4983ecD6B1c1596e8a5f976cf6CD9e3")},
{"wstETH", "StandardBridge", common.HexToAddress("0x65321bf24210b81500230dCEce14Faa70a9f50a7")},
},
}
func BridgesByChainID(chainID *big.Int, client bind.ContractBackend, addrs services.AddressManager) (map[string]Bridge, error) {
l1SBAddr, _ := addrs.L1StandardBridge()
allCfgs := []*implConfig{
{"Standard", "StandardBridge", l1SBAddr},
{"ETH", "ETHBridge", l1SBAddr},
}
allCfgs = append(allCfgs, customBridgeCfgs[chainID.Uint64()]...)
bridges := make(map[string]Bridge)
for _, bridge := range allCfgs {
switch bridge.impl {
case "StandardBridge":
l1SB, err := bindings.NewL1StandardBridge(bridge.addr, client)
if err != nil {
return nil, err
}
standardBridge := &StandardBridge{
name: bridge.name,
address: bridge.addr,
contract: l1SB,
}
bridges[bridge.name] = standardBridge
case "ETHBridge":
l1SB, err := bindings.NewL1StandardBridge(bridge.addr, client)
if err != nil {
return nil, err
}
ethBridge := &EthBridge{
name: bridge.name,
address: bridge.addr,
contract: l1SB,
}
bridges[bridge.name] = ethBridge
default:
return nil, errors.New("unsupported bridge")
}
}
return bridges, nil
}
func StateCommitmentChainScanner(client bind.ContractFilterer, addrs services.AddressManager) (*scc.StateCommitmentChainFilterer, error) {
sccAddr, _ := addrs.StateCommitmentChain()
filter, err := scc.NewStateCommitmentChainFilterer(sccAddr, client)
if err != nil {
return nil, err
}
return filter, nil
}
indexer/services/l1/bridge/eth_bridge.go
View file @ 1636250a
......@@ -5,66 +5,43 @@ import (
"github.com/ethereum-optimism/optimism/indexer/db"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum-optimism/optimism/op-service/backoff"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
"github.com/ethereum/go-ethereum/common"
)
type EthBridge struct {
name string
ctx context.Context
address common.Address
client bind.ContractFilterer
filterer *bindings.L1StandardBridgeFilterer
contract *bindings.L1StandardBridge
}
func (e *EthBridge) Address() common.Address {
return e.address
}
func (e *EthBridge) GetDepositsByBlockRange(start, end uint64) (DepositsMap, error) {
func (e *EthBridge) GetDepositsByBlockRange(ctx context.Context, start, end uint64) (DepositsMap, error) {
depositsByBlockhash := make(DepositsMap)
iter, err := FilterETHDepositInitiatedWithRetry(e.ctx, e.filterer, &bind.FilterOpts{
opts := &bind.FilterOpts{
Context: ctx,
Start: start,
End: &end,
})
if err != nil {
logger.Error("Error fetching filter", "err", err)
}
for iter.Next() {
depositsByBlockhash[iter.Event.Raw.BlockHash] = append(
depositsByBlockhash[iter.Event.Raw.BlockHash], db.Deposit{
TxHash: iter.Event.Raw.TxHash,
FromAddress: iter.Event.From,
ToAddress: iter.Event.To,
Amount: iter.Event.Amount,
Data: iter.Event.ExtraData,
LogIndex: iter.Event.Raw.Index,
})
}
if err := iter.Error(); err != nil {
return nil, err
}
return depositsByBlockhash, nil
}
func (s *EthBridge) GetWithdrawalsByBlockRange(start, end uint64) (WithdrawalsMap, error) {
withdrawalsByBlockHash := make(WithdrawalsMap)
iter, err := FilterETHWithdrawalFinalizedWithRetry(s.ctx, s.filterer, &bind.FilterOpts{
Start: start,
End: &end,
var iter *bindings.L1StandardBridgeETHDepositInitiatedIterator
err := backoff.Do(3, backoff.Exponential(), func() error {
var err error
iter, err = e.contract.FilterETHDepositInitiated(opts, nil, nil)
return err
})
if err != nil {
logger.Error("Error fetching filter", "err", err)
return nil, err
}
defer iter.Close()
for iter.Next() {
withdrawalsByBlockHash[iter.Event.Raw.BlockHash] = append(
withdrawalsByBlockHash[iter.Event.Raw.BlockHash], db.Withdrawal{
depositsByBlockhash[iter.Event.Raw.BlockHash] = append(
depositsByBlockhash[iter.Event.Raw.BlockHash], db.Deposit{
TxHash: iter.Event.Raw.TxHash,
FromAddress: iter.Event.From,
ToAddress: iter.Event.To,
......@@ -73,11 +50,8 @@ func (s *EthBridge) GetWithdrawalsByBlockRange(start, end uint64) (WithdrawalsMa
LogIndex: iter.Event.Raw.Index,
})
}
if err := iter.Error(); err != nil {
return nil, err
}
return withdrawalsByBlockHash, nil
return depositsByBlockhash, iter.Error()
}
func (e *EthBridge) String() string {
......
indexer/services/l1/bridge/filter.go
View file @ 1636250a
......@@ -4,7 +4,7 @@ import (
"context"
"time"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum-optimism/optimism/indexer/bindings/legacy/scc"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
)
......@@ -12,61 +12,13 @@ import (
// calls.
var clientRetryInterval = 5 * time.Second
// FilterETHDepositInitiatedWithRetry retries the given func until it succeeds,
// FilterStateBatchAppendedWithRetry retries the given func until it succeeds,
// waiting for clientRetryInterval duration after every call.
func FilterETHDepositInitiatedWithRetry(ctx context.Context, filterer *bindings.L1StandardBridgeFilterer, opts *bind.FilterOpts) (*bindings.L1StandardBridgeETHDepositInitiatedIterator, error) {
func FilterStateBatchAppendedWithRetry(ctx context.Context, filterer *scc.StateCommitmentChainFilterer, opts *bind.FilterOpts) (*scc.StateCommitmentChainStateBatchAppendedIterator, error) {
for {
ctxt, cancel := context.WithTimeout(ctx, DefaultConnectionTimeout)
opts.Context = ctxt
res, err := filterer.FilterETHDepositInitiated(opts, nil, nil)
cancel()
if err == nil {
return res, nil
}
logger.Error("Error fetching filter", "err", err)
time.Sleep(clientRetryInterval)
}
}
// FilterERC20DepositInitiatedWithRetry retries the given func until it succeeds,
// waiting for clientRetryInterval duration after every call.
func FilterERC20DepositInitiatedWithRetry(ctx context.Context, filterer *bindings.L1StandardBridgeFilterer, opts *bind.FilterOpts) (*bindings.L1StandardBridgeERC20DepositInitiatedIterator, error) {
for {
ctxt, cancel := context.WithTimeout(ctx, DefaultConnectionTimeout)
opts.Context = ctxt
res, err := filterer.FilterERC20DepositInitiated(opts, nil, nil, nil)
cancel()
if err == nil {
return res, nil
}
logger.Error("Error fetching filter", "err", err)
time.Sleep(clientRetryInterval)
}
}
// FilterETHWithdrawalFinalizedWithRetry retries the given func until it succeeds,
// waiting for clientRetryInterval duration after every call.
func FilterETHWithdrawalFinalizedWithRetry(ctx context.Context, filterer *bindings.L1StandardBridgeFilterer, opts *bind.FilterOpts) (*bindings.L1StandardBridgeETHWithdrawalFinalizedIterator, error) {
for {
ctxt, cancel := context.WithTimeout(ctx, DefaultConnectionTimeout)
opts.Context = ctxt
res, err := filterer.FilterETHWithdrawalFinalized(opts, nil, nil)
cancel()
if err == nil {
return res, nil
}
logger.Error("Error fetching filter", "err", err)
time.Sleep(clientRetryInterval)
}
}
// FilterERC20WithdrawalFinalizedWithRetry retries the given func until it succeeds,
// waiting for clientRetryInterval duration after every call.
func FilterERC20WithdrawalFinalizedWithRetry(ctx context.Context, filterer *bindings.L1StandardBridgeFilterer, opts *bind.FilterOpts) (*bindings.L1StandardBridgeERC20WithdrawalFinalizedIterator, error) {
for {
ctxt, cancel := context.WithTimeout(ctx, DefaultConnectionTimeout)
opts.Context = ctxt
res, err := filterer.FilterERC20WithdrawalFinalized(opts, nil, nil, nil)
res, err := filterer.FilterStateBatchAppended(opts, nil)
cancel()
if err == nil {
return res, nil
......
indexer/services/l1/bridge/portal.go 0 → 100644
View file @ 1636250a
package bridge
import (
"context"
"github.com/ethereum-optimism/optimism/indexer/db"
"github.com/ethereum-optimism/optimism/indexer/services"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum-optimism/optimism/op-service/backoff"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
"github.com/ethereum/go-ethereum/common"
)
type Portal struct {
address common.Address
contract *bindings.OptimismPortal
}
func NewPortal(addrs services.AddressManager) *Portal {
address, contract := addrs.OptimismPortal()
return &Portal{
address: address,
contract: contract,
}
}
func (p *Portal) Address() common.Address {
return p.address
}
func (p *Portal) GetFinalizedWithdrawalsByBlockRange(ctx context.Context, start, end uint64) (FinalizedWithdrawalsMap, error) {
wdsByBlockHash := make(FinalizedWithdrawalsMap)
opts := &bind.FilterOpts{
Context: ctx,
Start: start,
End: &end,
}
var iter *bindings.OptimismPortalWithdrawalFinalizedIterator
err := backoff.Do(3, backoff.Exponential(), func() error {
var err error
iter, err = p.contract.FilterWithdrawalFinalized(opts, nil)
return err
})
if err != nil {
return nil, err
}
defer iter.Close()
for iter.Next() {
wdsByBlockHash[iter.Event.Raw.BlockHash] = append(
wdsByBlockHash[iter.Event.Raw.BlockHash], db.FinalizedWithdrawal{
TxHash: iter.Event.Raw.TxHash,
WithdrawalHash: iter.Event.WithdrawalHash,
Success: iter.Event.Success,
LogIndex: iter.Event.Raw.Index,
},
)
}
return wdsByBlockHash, iter.Error()
}
indexer/services/l1/bridge/standard_bridge.go
View file @ 1636250a
......@@ -5,68 +5,43 @@ import (
"github.com/ethereum-optimism/optimism/indexer/db"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum-optimism/optimism/op-service/backoff"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
"github.com/ethereum/go-ethereum/common"
)
type StandardBridge struct {
name string
ctx context.Context
address common.Address
client bind.ContractFilterer
filterer *bindings.L1StandardBridgeFilterer
contract *bindings.L1StandardBridge
}
func (s *StandardBridge) Address() common.Address {
return s.address
}
func (s *StandardBridge) GetDepositsByBlockRange(start, end uint64) (DepositsMap, error) {
func (s *StandardBridge) GetDepositsByBlockRange(ctx context.Context, start, end uint64) (DepositsMap, error) {
depositsByBlockhash := make(DepositsMap)
iter, err := FilterERC20DepositInitiatedWithRetry(s.ctx, s.filterer, &bind.FilterOpts{
opts := &bind.FilterOpts{
Context: ctx,
Start: start,
End: &end,
})
if err != nil {
logger.Error("Error fetching filter", "err", err)
}
for iter.Next() {
depositsByBlockhash[iter.Event.Raw.BlockHash] = append(
depositsByBlockhash[iter.Event.Raw.BlockHash], db.Deposit{
TxHash: iter.Event.Raw.TxHash,
L1Token: iter.Event.L1Token,
L2Token: iter.Event.L2Token,
FromAddress: iter.Event.From,
ToAddress: iter.Event.To,
Amount: iter.Event.Amount,
Data: iter.Event.ExtraData,
LogIndex: iter.Event.Raw.Index,
})
}
if err := iter.Error(); err != nil {
return nil, err
}
return depositsByBlockhash, nil
}
func (s *StandardBridge) GetWithdrawalsByBlockRange(start, end uint64) (WithdrawalsMap, error) {
withdrawalsByBlockHash := make(WithdrawalsMap)
iter, err := FilterERC20WithdrawalFinalizedWithRetry(s.ctx, s.filterer, &bind.FilterOpts{
Start: start,
End: &end,
var iter *bindings.L1StandardBridgeERC20DepositInitiatedIterator
err := backoff.Do(3, backoff.Exponential(), func() error {
var err error
iter, err = s.contract.FilterERC20DepositInitiated(opts, nil, nil, nil)
return err
})
if err != nil {
logger.Error("Error fetching filter", "err", err)
return nil, err
}
defer iter.Close()
for iter.Next() {
withdrawalsByBlockHash[iter.Event.Raw.BlockHash] = append(
withdrawalsByBlockHash[iter.Event.Raw.BlockHash], db.Withdrawal{
depositsByBlockhash[iter.Event.Raw.BlockHash] = append(
depositsByBlockhash[iter.Event.Raw.BlockHash], db.Deposit{
TxHash: iter.Event.Raw.TxHash,
L1Token: iter.Event.L1Token,
L2Token: iter.Event.L2Token,
......@@ -77,11 +52,8 @@ func (s *StandardBridge) GetWithdrawalsByBlockRange(start, end uint64) (Withdraw
LogIndex: iter.Event.Raw.Index,
})
}
if err := iter.Error(); err != nil {
return nil, err
}
return withdrawalsByBlockHash, nil
return depositsByBlockhash, iter.Error()
}
func (s *StandardBridge) String() string {
......
indexer/services/l1/confirmed_headers.go
View file @ 1636250a
......@@ -18,7 +18,6 @@ import (
const (
DefaultConnectionTimeout = 30 * time.Second
DefaultConfDepth uint64 = 20
DefaultMaxBatchSize = 100
)
......
indexer/services/l1/query.go
View file @ 1636250a
package l1
import (
"github.com/ethereum/go-ethereum/ethclient"
"context"
"github.com/ethereum-optimism/optimism/indexer/bindings/legacy/scc"
"github.com/ethereum-optimism/optimism/indexer/db"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum-optimism/optimism/indexer/services/l1/bridge"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
"github.com/ethereum/go-ethereum/common"
)
func QueryERC20(address common.Address, client *ethclient.Client) (*db.Token, error) {
contract, err := bindings.NewERC20(address, client)
if err != nil {
return nil, err
}
name, err := contract.Name(&bind.CallOpts{})
if err != nil {
return nil, err
}
func QueryStateBatches(filterer *scc.StateCommitmentChainFilterer, startHeight, endHeight uint64, ctx context.Context) (map[common.Hash][]db.StateBatch, error) {
batches := make(map[common.Hash][]db.StateBatch)
symbol, err := contract.Symbol(&bind.CallOpts{})
iter, err := bridge.FilterStateBatchAppendedWithRetry(ctx, filterer, &bind.FilterOpts{
Start: startHeight,
End: &endHeight,
})
if err != nil {
return nil, err
}
decimals, err := contract.Decimals(&bind.CallOpts{})
if err != nil {
return nil, err
defer iter.Close()
for iter.Next() {
batches[iter.Event.Raw.BlockHash] = append(
batches[iter.Event.Raw.BlockHash], db.StateBatch{
Index: iter.Event.BatchIndex,
Root: iter.Event.BatchRoot,
Size: iter.Event.BatchSize,
PrevTotal: iter.Event.PrevTotalElements,
ExtraData: iter.Event.ExtraData,
BlockHash: iter.Event.Raw.BlockHash,
})
}
return &db.Token{
Name: name,
Symbol: symbol,
Decimals: decimals,
}, nil
return batches, iter.Error()
}
indexer/services/l1/service.go
View file @ 1636250a
......@@ -11,7 +11,10 @@ import (
"sync/atomic"
"time"
"github.com/ethereum-optimism/optimism/indexer/bindings/legacy/scc"
"github.com/ethereum-optimism/optimism/indexer/metrics"
"github.com/ethereum-optimism/optimism/indexer/services"
"github.com/ethereum-optimism/optimism/indexer/services/query"
"github.com/prometheus/client_golang/prometheus"
"github.com/ethereum-optimism/optimism/indexer/server"
......@@ -36,28 +39,8 @@ var errNoChainID = errors.New("no chain id provided")
var errNoNewBlocks = errors.New("no new blocks")
// clientRetryInterval is the interval to wait between retrying client API
// calls.
var clientRetryInterval = 5 * time.Second
var ZeroAddress common.Address
// HeaderByNumberWithRetry retries the given func until it succeeds, waiting
// for clientRetryInterval duration after every call.
func HeaderByNumberWithRetry(ctx context.Context,
client *ethclient.Client) (*types.Header, error) {
for {
res, err := client.HeaderByNumber(ctx, nil)
switch err {
case nil:
return res, err
default:
log.Error("Error fetching header", "err", err)
}
time.Sleep(clientRetryInterval)
}
}
// Driver is an interface for indexing deposits from l1.
type Driver interface {
// Name is an identifier used to prefix logs for a particular service.
......@@ -70,11 +53,12 @@ type ServiceConfig struct {
L1Client *ethclient.Client
RawL1Client *rpc.Client
ChainID *big.Int
AddressManager services.AddressManager
ConfDepth uint64
MaxHeaderBatchSize uint64
StartBlockNumber uint64
StartBlockHash string
DB *db.Database
Bedrock bool
}
type Service struct {
......@@ -83,11 +67,14 @@ type Service struct {
cancel func()
bridges map[string]bridge.Bridge
portal *bridge.Portal
batchScanner *scc.StateCommitmentChainFilterer
latestHeader uint64
headerSelector *ConfirmedHeaderSelector
metrics *metrics.Metrics
tokenCache map[common.Address]*db.Token
isBedrock bool
wg sync.WaitGroup
}
......@@ -113,11 +100,24 @@ func NewService(cfg ServiceConfig) (*Service, error) {
return nil, fmt.Errorf("chain ID configured with %d but got %d", cfg.ChainID, chainID)
}
bridges, err := bridge.BridgesByChainID(cfg.ChainID, cfg.L1Client, ctx)
bridges, err := bridge.BridgesByChainID(cfg.ChainID, cfg.L1Client, cfg.AddressManager)
if err != nil {
cancel()
return nil, err
}
var portal *bridge.Portal
var batchScanner *scc.StateCommitmentChainFilterer
if cfg.Bedrock {
portal = bridge.NewPortal(cfg.AddressManager)
} else {
batchScanner, err = bridge.StateCommitmentChainScanner(cfg.L1Client, cfg.AddressManager)
if err != nil {
cancel()
return nil, err
}
}
logger.Info("Scanning bridges for deposits", "bridges", bridges)
confirmedHeaderSelector, err := NewConfirmedHeaderSelector(HeaderSelectorConfig{
......@@ -130,22 +130,29 @@ func NewService(cfg ServiceConfig) (*Service, error) {
return nil, err
}
return &Service{
service := &Service{
cfg: cfg,
ctx: ctx,
cancel: cancel,
portal: portal,
bridges: bridges,
batchScanner: batchScanner,
headerSelector: confirmedHeaderSelector,
metrics: cfg.Metrics,
tokenCache: map[common.Address]*db.Token{
ZeroAddress: db.ETHL1Token,
},
}, nil
isBedrock: cfg.Bedrock,
}
service.wg.Add(1)
return service, nil
}
func (s *Service) Loop(ctx context.Context) {
func (s *Service) loop() {
defer s.wg.Done()
for {
err := s.catchUp(ctx)
err := s.catchUp()
if err == nil {
break
}
......@@ -159,10 +166,18 @@ func (s *Service) Loop(ctx context.Context) {
}
newHeads := make(chan *types.Header, 1000)
go s.subscribeNewHeads(ctx, newHeads)
tick := time.NewTicker(5 * time.Second)
defer tick.Stop()
for {
select {
case <-tick.C:
header, err := query.HeaderByNumberWithRetry(s.ctx, s.cfg.L1Client)
if err != nil {
logger.Error("error fetching header by number", "err", err)
continue
}
newHeads <- header
case header := <-newHeads:
if header == nil {
break
......@@ -180,6 +195,7 @@ func (s *Service) Loop(ctx context.Context) {
}
}
case <-s.ctx.Done():
logger.Info("service stopped")
return
}
}
......@@ -188,7 +204,6 @@ func (s *Service) Loop(ctx context.Context) {
func (s *Service) Update(newHeader *types.Header) error {
var lowest = db.BlockLocator{
Number: s.cfg.StartBlockNumber,
Hash: common.HexToHash(s.cfg.StartBlockHash),
}
highestConfirmed, err := s.cfg.DB.GetHighestL1Block()
if err != nil {
......@@ -213,7 +228,7 @@ func (s *Service) Update(newHeader *types.Header) error {
return nil
}
if lowest.Hash != headers[0].ParentHash {
if lowest.Number > 0 && lowest.Hash != headers[0].ParentHash {
logger.Error("Parent hash does not connect to ",
"block", headers[0].Number.Uint64(), "hash", headers[0].Hash,
"lowest_block", lowest.Number, "hash", lowest.Hash)
......@@ -223,7 +238,6 @@ func (s *Service) Update(newHeader *types.Header) error {
startHeight := headers[0].Number.Uint64()
endHeight := headers[len(headers)-1].Number.Uint64()
depositsByBlockHash := make(map[common.Hash][]db.Deposit)
withdrawalsByBlockHash := make(map[common.Hash][]db.Withdrawal)
start := prometheus.NewTimer(s.metrics.UpdateDuration.WithLabelValues("l1"))
defer func() {
......@@ -232,27 +246,27 @@ func (s *Service) Update(newHeader *types.Header) error {
}()
bridgeDepositsCh := make(chan bridge.DepositsMap, len(s.bridges))
bridgeWdsCh := make(chan bridge.WithdrawalsMap, len(s.bridges))
errCh := make(chan error, len(s.bridges))
finalizedWithdrawalsCh := make(chan bridge.FinalizedWithdrawalsMap, 1)
errCh := make(chan error, len(s.bridges)+1)
for _, bridgeImpl := range s.bridges {
go func(b bridge.Bridge) {
deposits, err := b.GetDepositsByBlockRange(startHeight, endHeight)
deposits, err := b.GetDepositsByBlockRange(s.ctx, startHeight, endHeight)
if err != nil {
errCh <- err
return
}
bridgeDepositsCh <- deposits
}(bridgeImpl)
go func(b bridge.Bridge) {
withdrawals, err := b.GetWithdrawalsByBlockRange(startHeight, endHeight)
}
go func() {
finalizedWithdrawals, err := s.portal.GetFinalizedWithdrawalsByBlockRange(s.ctx, startHeight, endHeight)
if err != nil {
errCh <- err
return
}
bridgeWdsCh <- withdrawals
}(bridgeImpl)
}
finalizedWithdrawalsCh <- finalizedWithdrawals
}()
var receives int
for {
......@@ -260,40 +274,44 @@ func (s *Service) Update(newHeader *types.Header) error {
case bridgeDeposits := <-bridgeDepositsCh:
for blockHash, deposits := range bridgeDeposits {
for _, deposit := range deposits {
if err := s.cacheToken(deposit.L1Token); err != nil {
if err := s.cacheToken(deposit); err != nil {
logger.Warn("error caching token", "err", err)
}
}
depositsByBlockHash[blockHash] = append(depositsByBlockHash[blockHash], deposits...)
}
case bridgeWithdrawals := <-bridgeWdsCh:
for blockHash, withdrawals := range bridgeWithdrawals {
for _, withdrawal := range withdrawals {
if err := s.cacheToken(withdrawal.L1Token); err != nil {
logger.Warn("error caching token", "err", err)
}
}
withdrawalsByBlockHash[blockHash] = append(withdrawalsByBlockHash[blockHash], withdrawals...)
}
case err := <-errCh:
return err
}
receives++
if receives == 2*len(s.bridges) {
if receives == len(s.bridges) {
break
}
}
finalizedWithdrawalsByBlockHash := <-finalizedWithdrawalsCh
var stateBatches map[common.Hash][]db.StateBatch
if !s.isBedrock {
stateBatches, err = QueryStateBatches(s.batchScanner, startHeight, endHeight, s.ctx)
if err != nil {
logger.Error("Error querying state batches", "err", err)
return err
}
}
for i, header := range headers {
blockHash := header.Hash
number := header.Number.Uint64()
deposits := depositsByBlockHash[blockHash]
withdrawals := withdrawalsByBlockHash[blockHash]
batches := stateBatches[blockHash]
finalizedWds := finalizedWithdrawalsByBlockHash[blockHash]
if len(deposits) == 0 && len(withdrawals) == 0 && i != len(headers)-1 {
// Always record block data in the last block
// in the list of headers
if len(deposits) == 0 && len(batches) == 0 && len(finalizedWds) == 0 && i != len(headers)-1 {
continue
}
......@@ -303,6 +321,7 @@ func (s *Service) Update(newHeader *types.Header) error {
Number: number,
Timestamp: header.Time,
Deposits: deposits,
FinalizedWithdrawals: finalizedWds,
}
err := s.cfg.DB.AddIndexedL1Block(block)
......@@ -310,24 +329,35 @@ func (s *Service) Update(newHeader *types.Header) error {
logger.Error(
"Unable to import ",
"block", number,
"hash", blockHash,
"err", err,
"hash", blockHash, "err", err,
"block", block,
)
return err
}
err = s.cfg.DB.AddStateBatch(batches)
if err != nil {
logger.Error(
"Unable to import state append batch",
"block", number,
"hash", blockHash, "err", err,
"block", block,
)
return err
}
s.metrics.RecordStateBatches(len(batches))
logger.Debug("Imported ",
"block", number, "hash", blockHash, "deposits", len(block.Deposits))
for _, deposit := range block.Deposits {
token := s.tokenCache[deposit.L2Token]
token := s.tokenCache[deposit.L1Token]
logger.Info(
"indexed deposit ",
"indexed deposit",
"tx_hash", deposit.TxHash,
"symbol", token.Symbol,
"amount", deposit.Amount,
)
s.metrics.RecordDeposit(deposit.L2Token)
s.metrics.RecordDeposit(deposit.L1Token)
}
}
......@@ -382,8 +412,8 @@ func (s *Service) GetDeposits(w http.ResponseWriter, r *http.Request) {
}
page := db.PaginationParam{
Limit: uint64(limit),
Offset: uint64(offset),
Limit: limit,
Offset: offset,
}
deposits, err := s.cfg.DB.GetDepositsByAddress(common.HexToAddress(vars["address"]), page)
......@@ -395,25 +425,8 @@ func (s *Service) GetDeposits(w http.ResponseWriter, r *http.Request) {
server.RespondWithJSON(w, http.StatusOK, deposits)
}
func (s *Service) subscribeNewHeads(ctx context.Context, heads chan *types.Header) {
tick := time.NewTicker(5 * time.Second)
for {
select {
case <-tick.C:
header, err := HeaderByNumberWithRetry(ctx, s.cfg.L1Client)
if err != nil {
logger.Error("error fetching header by number", "err", err)
}
heads <- header
case <-ctx.Done():
return
}
}
}
func (s *Service) catchUp(ctx context.Context) error {
realHead, err := HeaderByNumberWithRetry(ctx, s.cfg.L1Client)
func (s *Service) catchUp() error {
realHead, err := query.HeaderByNumberWithRetry(s.ctx, s.cfg.L1Client)
if err != nil {
return err
}
......@@ -437,8 +450,8 @@ func (s *Service) catchUp(ctx context.Context) error {
for realHeadNum-s.cfg.ConfDepth > currHeadNum+s.cfg.MaxHeaderBatchSize {
select {
case <-ctx.Done():
return context.Canceled
case <-s.ctx.Done():
return s.ctx.Err()
default:
if err := s.Update(realHead); err != nil && err != errNoNewBlocks {
return err
......@@ -456,33 +469,33 @@ func (s *Service) catchUp(ctx context.Context) error {
return nil
}
func (s *Service) cacheToken(address common.Address) error {
if s.tokenCache[address] != nil {
func (s *Service) cacheToken(deposit db.Deposit) error {
if s.tokenCache[deposit.L1Token] != nil {
return nil
}
token, err := s.cfg.DB.GetL1TokenByAddress(address.String())
token, err := s.cfg.DB.GetL1TokenByAddress(deposit.L1Token.String())
if err != nil {
return err
}
if token != nil {
s.metrics.IncL1CachedTokensCount()
s.tokenCache[address] = token
s.tokenCache[deposit.L1Token] = token
return nil
}
token, err = QueryERC20(address, s.cfg.L1Client)
token, err = query.NewERC20(deposit.L1Token, s.cfg.L1Client)
if err != nil {
logger.Error("Error querying ERC20 token details",
"l1_token", address.String(), "err", err)
"l1_token", deposit.L1Token.String(), "err", err)
token = &db.Token{
Address: address.String(),
Address: deposit.L1Token.String(),
}
}
if err := s.cfg.DB.AddL1Token(address.String(), token); err != nil {
if err := s.cfg.DB.AddL1Token(deposit.L1Token.String(), token); err != nil {
return err
}
s.tokenCache[address] = token
s.tokenCache[deposit.L1Token] = token
s.metrics.IncL1CachedTokensCount()
return nil
}
......@@ -491,16 +504,11 @@ func (s *Service) Start() error {
if s.cfg.ChainID == nil {
return errNoChainID
}
s.wg.Add(1)
go s.Loop(s.ctx)
go s.loop()
return nil
}
func (s *Service) Stop() {
s.cancel()
s.wg.Wait()
err := s.cfg.DB.Close()
if err != nil {
logger.Error("Error closing db", "err", err)
}
}
indexer/services/l2/bridge/bridge.go
View file @ 1636250a
......@@ -7,75 +7,76 @@ import (
"github.com/ethereum-optimism/optimism/indexer/db"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
"github.com/ethereum/go-ethereum/ethclient"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
"github.com/ethereum/go-ethereum/common"
)
type DepositsMap map[common.Hash][]db.Deposit // Finalizations
type WithdrawalsMap map[common.Hash][]db.Withdrawal
type Bridge interface {
Address() common.Address
GetDepositsByBlockRange(uint64, uint64) (DepositsMap, error)
GetWithdrawalsByBlockRange(uint64, uint64) (WithdrawalsMap, error)
GetWithdrawalsByBlockRange(context.Context, uint64, uint64) (WithdrawalsMap, error)
String() string
}
type implConfig struct {
name string
impl string
addr string
addr common.Address
}
var defaultBridgeCfgs = map[uint64][]*implConfig{
// Devnet
901: {
{"Standard", "StandardBridge", L2StandardBridgeAddr},
},
// Goerli Alpha Testnet
28528: {
{"Standard", "StandardBridge", L2StandardBridgeAddr},
},
var defaultBridgeCfgs = []*implConfig{
{"Standard", "StandardBridge", predeploys.L2StandardBridgeAddr},
}
var customBridgeCfgs = map[uint64][]*implConfig{
// Mainnet
10: {
{"BitBTC", StandardBridgeImpl, "0x158F513096923fF2d3aab2BcF4478536de6725e2"},
{"BitBTC", StandardBridgeImpl, common.HexToAddress("0x158F513096923fF2d3aab2BcF4478536de6725e2")},
//{"DAI", "DAIBridge", "0x467194771dAe2967Aef3ECbEDD3Bf9a310C76C65"},
{"wstETH", StandardBridgeImpl, common.HexToAddress("0x8E01013243a96601a86eb3153F0d9Fa4fbFb6957")},
},
// Kovan
69: {
{"BitBTC", StandardBridgeImpl, "0x0CFb46528a7002a7D8877a5F7a69b9AaF1A9058e"},
{"USX", StandardBridgeImpl, "0xB4d37826b14Cd3CB7257A2A5094507d701fe715f"},
{"BitBTC", StandardBridgeImpl, common.HexToAddress("0x0CFb46528a7002a7D8877a5F7a69b9AaF1A9058e")},
{"USX", StandardBridgeImpl, common.HexToAddress("0xB4d37826b14Cd3CB7257A2A5094507d701fe715f")},
{"wstETH", StandardBridgeImpl, common.HexToAddress("0x2E34e7d705AfaC3C4665b6feF31Aa394A1c81c92")},
//{"DAI", " DAIBridge", "0x467194771dAe2967Aef3ECbEDD3Bf9a310C76C65"},
},
}
func BridgesByChainID(chainID *big.Int, client bind.ContractFilterer, ctx context.Context) (map[string]Bridge, error) {
func BridgesByChainID(chainID *big.Int, client *ethclient.Client, isBedrock bool) (map[string]Bridge, error) {
allCfgs := make([]*implConfig, 0)
allCfgs = append(allCfgs, defaultBridgeCfgs[chainID.Uint64()]...)
allCfgs = append(allCfgs, defaultBridgeCfgs...)
allCfgs = append(allCfgs, customBridgeCfgs[chainID.Uint64()]...)
var l2L1MP *bindings.L2ToL1MessagePasser
var err error
if isBedrock {
l2L1MP, err = bindings.NewL2ToL1MessagePasser(predeploys.L2ToL1MessagePasserAddr, client)
if err != nil {
return nil, err
}
}
bridges := make(map[string]Bridge)
for _, bridge := range allCfgs {
switch bridge.impl {
case "StandardBridge":
l2StandardBridgeAddress := common.HexToAddress(bridge.addr)
l2StandardBridgeFilter, err := bindings.NewL2StandardBridgeFilterer(l2StandardBridgeAddress, client)
l2SB, err := bindings.NewL2StandardBridge(bridge.addr, client)
if err != nil {
return nil, err
}
standardBridge := &StandardBridge{
bridges[bridge.name] = &StandardBridge{
name: bridge.name,
ctx: ctx,
address: l2StandardBridgeAddress,
address: bridge.addr,
client: client,
filterer: l2StandardBridgeFilter,
l2SB: l2SB,
l2L1MP: l2L1MP,
isBedrock: isBedrock,
}
bridges[bridge.name] = standardBridge
default:
return nil, errors.New("unsupported bridge")
}
......
indexer/services/l2/bridge/filter.go deleted 100644 → 0
View file @ 01c947c0
package bridge
import (
"context"
"time"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
)
// clientRetryInterval is the interval to wait between retrying client API
// calls.
var clientRetryInterval = 5 * time.Second
// FilterWithdrawalInitiatedWithRetry retries the given func until it succeeds,
// waiting for clientRetryInterval duration after every call.
func FilterWithdrawalInitiatedWithRetry(ctx context.Context, filterer *bindings.L2StandardBridgeFilterer, opts *bind.FilterOpts) (*bindings.L2StandardBridgeWithdrawalInitiatedIterator, error) {
for {
ctxt, cancel := context.WithTimeout(ctx, DefaultConnectionTimeout)
opts.Context = ctxt
res, err := filterer.FilterWithdrawalInitiated(opts, nil, nil, nil)
cancel()
if err == nil {
return res, nil
}
logger.Error("Error fetching filter", "err", err)
time.Sleep(clientRetryInterval)
}
}
// FilterDepositFinalizedWithRetry retries the given func until it succeeds,
// waiting for clientRetryInterval duration after every call.
func FilterDepositFinalizedWithRetry(ctx context.Context, filterer *bindings.L2StandardBridgeFilterer, opts *bind.FilterOpts) (*bindings.L2StandardBridgeDepositFinalizedIterator, error) {
for {
ctxt, cancel := context.WithTimeout(ctx, DefaultConnectionTimeout)
opts.Context = ctxt
res, err := filterer.FilterDepositFinalized(opts, nil, nil, nil)
cancel()
if err == nil {
return res, nil
}
logger.Error("Error fetching filter", "err", err)
time.Sleep(clientRetryInterval)
}
}
indexer/services/l2/bridge/standard_bridge.go
View file @ 1636250a
......@@ -5,6 +5,10 @@ import (
"github.com/ethereum-optimism/optimism/indexer/db"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum-optimism/optimism/op-node/withdrawals"
"github.com/ethereum-optimism/optimism/op-service/backoff"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/ethclient"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
"github.com/ethereum/go-ethereum/common"
......@@ -12,76 +16,101 @@ import (
type StandardBridge struct {
name string
ctx context.Context
address common.Address
client bind.ContractFilterer
filterer *bindings.L2StandardBridgeFilterer
client *ethclient.Client
l2SB *bindings.L2StandardBridge
l2L1MP *bindings.L2ToL1MessagePasser
isBedrock bool
}
func (s *StandardBridge) Address() common.Address {
return s.address
}
func (s *StandardBridge) GetDepositsByBlockRange(start, end uint64) (DepositsMap, error) {
depositsByBlockhash := make(DepositsMap)
iter, err := FilterDepositFinalizedWithRetry(s.ctx, s.filterer, &bind.FilterOpts{
func (s *StandardBridge) GetWithdrawalsByBlockRange(ctx context.Context, start, end uint64) (WithdrawalsMap, error) {
withdrawalsByBlockhash := make(map[common.Hash][]db.Withdrawal)
opts := &bind.FilterOpts{
Context: ctx,
Start: start,
End: &end,
}
var iter *bindings.L2StandardBridgeWithdrawalInitiatedIterator
err := backoff.Do(3, backoff.Exponential(), func() error {
var err error
iter, err = s.l2SB.FilterWithdrawalInitiated(opts, nil, nil, nil)
return err
})
if err != nil {
logger.Error("Error fetching filter", "err", err)
return nil, err
}
receipts := make(map[common.Hash]*types.Receipt)
defer iter.Close()
for iter.Next() {
depositsByBlockhash[iter.Event.Raw.BlockHash] = append(
depositsByBlockhash[iter.Event.Raw.BlockHash], db.Deposit{
TxHash: iter.Event.Raw.TxHash,
L1Token: iter.Event.L1Token,
L2Token: iter.Event.L2Token,
FromAddress: iter.Event.From,
ToAddress: iter.Event.To,
Amount: iter.Event.Amount,
Data: iter.Event.ExtraData,
LogIndex: iter.Event.Raw.Index,
})
}
if err := iter.Error(); err != nil {
ev := iter.Event
if s.isBedrock {
receipt := receipts[ev.Raw.TxHash]
if receipt == nil {
receipt, err = s.client.TransactionReceipt(ctx, ev.Raw.TxHash)
if err != nil {
return nil, err
}
receipts[ev.Raw.TxHash] = receipt
}
return depositsByBlockhash, nil
}
var withdrawalInitiated *bindings.L2ToL1MessagePasserMessagePassed
for _, eLog := range receipt.Logs {
if len(eLog.Topics) == 0 || eLog.Topics[0] != withdrawals.MessagePassedTopic {
continue
}
func (s *StandardBridge) GetWithdrawalsByBlockRange(start, end uint64) (WithdrawalsMap, error) {
withdrawalsByBlockhash := make(map[common.Hash][]db.Withdrawal)
if withdrawalInitiated != nil {
logger.Warn("detected multiple withdrawal initiated events! ignoring", "tx_hash", ev.Raw.TxHash)
continue
}
iter, err := FilterWithdrawalInitiatedWithRetry(s.ctx, s.filterer, &bind.FilterOpts{
Start: start,
End: &end,
})
withdrawalInitiated, err = s.l2L1MP.ParseMessagePassed(*eLog)
if err != nil {
logger.Error("Error fetching filter", "err", err)
return nil, err
}
for iter.Next() {
withdrawalsByBlockhash[iter.Event.Raw.BlockHash] = append(
withdrawalsByBlockhash[iter.Event.Raw.BlockHash], db.Withdrawal{
TxHash: iter.Event.Raw.TxHash,
L1Token: iter.Event.L1Token,
L2Token: iter.Event.L2Token,
FromAddress: iter.Event.From,
ToAddress: iter.Event.To,
Amount: iter.Event.Amount,
Data: iter.Event.ExtraData,
LogIndex: iter.Event.Raw.Index,
})
}
if err := iter.Error(); err != nil {
hash, err := withdrawals.WithdrawalHash(withdrawalInitiated)
if err != nil {
return nil, err
}
return withdrawalsByBlockhash, nil
withdrawalsByBlockhash[ev.Raw.BlockHash] = append(
withdrawalsByBlockhash[ev.Raw.BlockHash], db.Withdrawal{
TxHash: ev.Raw.TxHash,
L1Token: ev.L1Token,
L2Token: ev.L2Token,
FromAddress: ev.From,
ToAddress: ev.To,
Amount: ev.Amount,
Data: ev.ExtraData,
LogIndex: ev.Raw.Index,
BedrockHash: &hash,
},
)
} else {
withdrawalsByBlockhash[ev.Raw.BlockHash] = append(
withdrawalsByBlockhash[ev.Raw.BlockHash], db.Withdrawal{
TxHash: ev.Raw.TxHash,
L1Token: ev.L1Token,
L2Token: ev.L2Token,
FromAddress: ev.From,
ToAddress: ev.To,
Amount: ev.Amount,
Data: ev.ExtraData,
LogIndex: ev.Raw.Index,
},
)
}
}
return withdrawalsByBlockhash, iter.Error()
}
func (s *StandardBridge) String() string {
......
indexer/services/l2/confirmed_headers.go
View file @ 1636250a
......@@ -147,8 +147,7 @@ func (f *ConfirmedHeaderSelector) NewHead(
return headers, nil
}
func NewConfirmedHeaderSelector(cfg HeaderSelectorConfig) (*ConfirmedHeaderSelector,
error) {
func NewConfirmedHeaderSelector(cfg HeaderSelectorConfig) (*ConfirmedHeaderSelector, error) {
if cfg.ConfDepth == 0 {
return nil, errors.New("ConfDepth must be greater than zero")
}
......
indexer/services/l2/service.go
View file @ 1636250a
......@@ -12,6 +12,8 @@ import (
"github.com/ethereum-optimism/optimism/indexer/metrics"
"github.com/ethereum-optimism/optimism/indexer/server"
"github.com/ethereum-optimism/optimism/indexer/services/query"
"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
"github.com/prometheus/client_golang/prometheus"
"github.com/ethereum-optimism/optimism/indexer/db"
......@@ -38,37 +40,18 @@ var errWrongChainID = errors.New("wrong chain id provided")
var errNoNewBlocks = errors.New("no new blocks")
// clientRetryInterval is the interval to wait between retrying client API
// calls.
var clientRetryInterval = 5 * time.Second
// HeaderByNumberWithRetry retries the given func until it succeeds, waiting
// for clientRetryInterval duration after every call.
func HeaderByNumberWithRetry(ctx context.Context,
client *ethclient.Client) (*types.Header, error) {
for {
res, err := client.HeaderByNumber(ctx, nil)
switch err {
case nil:
return res, err
default:
log.Error("Error fetching header", "err", err)
}
time.Sleep(clientRetryInterval)
}
}
type ServiceConfig struct {
Context context.Context
Metrics *metrics.Metrics
L2RPC *rpc.Client
L2Client *ethclient.Client
ChainID *big.Int
ConfDepth uint64
MaxHeaderBatchSize uint64
StartBlockNumber uint64
StartBlockHash string
DB *db.Database
Bedrock bool
}
type Service struct {
......@@ -113,7 +96,7 @@ func NewService(cfg ServiceConfig) (*Service, error) {
cfg.ChainID = chainID
}
bridges, err := bridge.BridgesByChainID(cfg.ChainID, cfg.L2Client, ctx)
bridges, err := bridge.BridgesByChainID(cfg.ChainID, cfg.L2Client, cfg.Bedrock)
if err != nil {
cancel()
return nil, err
......@@ -131,7 +114,7 @@ func NewService(cfg ServiceConfig) (*Service, error) {
return nil, err
}
return &Service{
service := &Service{
cfg: cfg,
ctx: ctx,
cancel: cancel,
......@@ -139,14 +122,18 @@ func NewService(cfg ServiceConfig) (*Service, error) {
headerSelector: confirmedHeaderSelector,
metrics: cfg.Metrics,
tokenCache: map[common.Address]*db.Token{
db.ETHL2Address: db.ETHL1Token,
predeploys.LegacyERC20ETHAddr: db.ETHL1Token,
},
}, nil
}
service.wg.Add(1)
return service, nil
}
func (s *Service) Loop(ctx context.Context) {
func (s *Service) loop() {
defer s.wg.Done()
for {
err := s.catchUp(ctx)
err := s.catchUp()
if err == nil {
break
}
......@@ -160,10 +147,18 @@ func (s *Service) Loop(ctx context.Context) {
}
newHeads := make(chan *types.Header, 1000)
go s.subscribeNewHeads(ctx, newHeads)
tick := time.NewTicker(5 * time.Second)
defer tick.Stop()
for {
select {
case <-tick.C:
header, err := query.HeaderByNumberWithRetry(s.ctx, s.cfg.L2Client)
if err != nil {
logger.Error("error fetching header by number", "err", err)
continue
}
newHeads <- header
case header := <-newHeads:
logger.Info("Received new header", "header", header.Hash)
for {
......@@ -176,6 +171,7 @@ func (s *Service) Loop(ctx context.Context) {
}
}
case <-s.ctx.Done():
logger.Info("service stopped")
return
}
}
......@@ -184,7 +180,6 @@ func (s *Service) Loop(ctx context.Context) {
func (s *Service) Update(newHeader *types.Header) error {
var lowest = db.BlockLocator{
Number: s.cfg.StartBlockNumber,
Hash: common.HexToHash(s.cfg.StartBlockHash),
}
highestConfirmed, err := s.cfg.DB.GetHighestL2Block()
if err != nil {
......@@ -209,7 +204,7 @@ func (s *Service) Update(newHeader *types.Header) error {
return nil
}
if lowest.Hash != headers[0].ParentHash {
if lowest.Number > 0 && lowest.Hash != headers[0].ParentHash {
logger.Error("Parent hash does not connect to ",
"block", headers[0].Number.Uint64(), "hash", headers[0].Hash(),
"lowest_block", lowest.Number, "hash", lowest.Hash)
......@@ -218,7 +213,6 @@ func (s *Service) Update(newHeader *types.Header) error {
startHeight := headers[0].Number.Uint64()
endHeight := headers[len(headers)-1].Number.Uint64()
depositsByBlockHash := make(map[common.Hash][]db.Deposit)
withdrawalsByBlockHash := make(map[common.Hash][]db.Withdrawal)
start := prometheus.NewTimer(s.metrics.UpdateDuration.WithLabelValues("l2"))
......@@ -227,21 +221,12 @@ func (s *Service) Update(newHeader *types.Header) error {
logger.Info("updated index", "start_height", startHeight, "end_height", endHeight, "duration", dur)
}()
bridgeDepositsCh := make(chan bridge.DepositsMap, len(s.bridges))
bridgeWdsCh := make(chan bridge.WithdrawalsMap)
errCh := make(chan error, len(s.bridges))
for _, bridgeImpl := range s.bridges {
go func(b bridge.Bridge) {
deposits, err := b.GetDepositsByBlockRange(startHeight, endHeight)
if err != nil {
errCh <- err
return
}
bridgeDepositsCh <- deposits
}(bridgeImpl)
go func(b bridge.Bridge) {
wds, err := b.GetWithdrawalsByBlockRange(startHeight, endHeight)
wds, err := b.GetWithdrawalsByBlockRange(s.ctx, startHeight, endHeight)
if err != nil {
errCh <- err
return
......@@ -256,29 +241,19 @@ func (s *Service) Update(newHeader *types.Header) error {
case bridgeWds := <-bridgeWdsCh:
for blockHash, withdrawals := range bridgeWds {
for _, wd := range withdrawals {
if err := s.cacheToken(wd.L2Token); err != nil {
if err := s.cacheToken(wd); err != nil {
logger.Warn("error caching token", "err", err)
}
}
withdrawalsByBlockHash[blockHash] = append(withdrawalsByBlockHash[blockHash], withdrawals...)
}
case bridgeDeposits := <-bridgeDepositsCh:
for blockHash, deposits := range bridgeDeposits {
for _, deposit := range deposits {
if err := s.cacheToken(deposit.L2Token); err != nil {
logger.Warn("error caching token", "err", err)
}
}
depositsByBlockHash[blockHash] = append(depositsByBlockHash[blockHash], deposits...)
}
case err := <-errCh:
return err
}
receives++
if receives == 2*len(s.bridges) {
if receives == len(s.bridges) {
break
}
}
......@@ -356,8 +331,21 @@ func (s *Service) GetIndexerStatus(w http.ResponseWriter, r *http.Request) {
server.RespondWithJSON(w, http.StatusOK, status)
}
func (s *Service) GetWithdrawalStatus(w http.ResponseWriter, r *http.Request) {
// Temporary stub until rest of indexer is landed
func (s *Service) GetWithdrawalBatch(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
hash := vars["hash"]
if hash == "" {
server.RespondWithError(w, http.StatusBadRequest, "must specify a hash")
return
}
batch, err := s.cfg.DB.GetWithdrawalBatch(common.HexToHash(vars["hash"]))
if err != nil {
server.RespondWithError(w, http.StatusInternalServerError, err.Error())
return
}
server.RespondWithJSON(w, http.StatusOK, batch)
}
func (s *Service) GetWithdrawals(w http.ResponseWriter, r *http.Request) {
......@@ -380,12 +368,14 @@ func (s *Service) GetWithdrawals(w http.ResponseWriter, r *http.Request) {
return
}
finalizationState := db.ParseFinalizationState(r.URL.Query().Get("finalized"))
page := db.PaginationParam{
Limit: uint64(limit),
Offset: uint64(offset),
Limit: limit,
Offset: offset,
}
withdrawals, err := s.cfg.DB.GetWithdrawalsByAddress(common.HexToAddress(vars["address"]), page)
withdrawals, err := s.cfg.DB.GetWithdrawalsByAddress(common.HexToAddress(vars["address"]), page, finalizationState)
if err != nil {
server.RespondWithError(w, http.StatusInternalServerError, err.Error())
return
......@@ -394,25 +384,8 @@ func (s *Service) GetWithdrawals(w http.ResponseWriter, r *http.Request) {
server.RespondWithJSON(w, http.StatusOK, withdrawals)
}
func (s *Service) subscribeNewHeads(ctx context.Context, heads chan *types.Header) {
tick := time.NewTicker(5 * time.Second)
for {
select {
case <-tick.C:
header, err := HeaderByNumberWithRetry(ctx, s.cfg.L2Client)
if err != nil {
logger.Error("error fetching header by number", "err", err)
}
heads <- header
case <-ctx.Done():
return
}
}
}
func (s *Service) catchUp(ctx context.Context) error {
realHead, err := HeaderByNumberWithRetry(ctx, s.cfg.L2Client)
func (s *Service) catchUp() error {
realHead, err := query.HeaderByNumberWithRetry(s.ctx, s.cfg.L2Client)
if err != nil {
return err
}
......@@ -436,8 +409,8 @@ func (s *Service) catchUp(ctx context.Context) error {
for realHeadNum-s.cfg.ConfDepth > currHeadNum+s.cfg.MaxHeaderBatchSize {
select {
case <-ctx.Done():
return context.Canceled
case <-s.ctx.Done():
return s.ctx.Err()
default:
if err := s.Update(realHead); err != nil && err != errNoNewBlocks {
return err
......@@ -455,32 +428,32 @@ func (s *Service) catchUp(ctx context.Context) error {
return nil
}
func (s *Service) cacheToken(address common.Address) error {
if s.tokenCache[address] != nil {
func (s *Service) cacheToken(withdrawal db.Withdrawal) error {
if s.tokenCache[withdrawal.L2Token] != nil {
return nil
}
token, err := s.cfg.DB.GetL2TokenByAddress(address.String())
token, err := s.cfg.DB.GetL2TokenByAddress(withdrawal.L2Token.String())
if err != nil {
return err
}
if token != nil {
s.metrics.IncL2CachedTokensCount()
s.tokenCache[address] = token
s.tokenCache[withdrawal.L2Token] = token
return nil
}
token, err = QueryERC20(address, s.cfg.L2Client)
token, err = query.NewERC20(withdrawal.L2Token, s.cfg.L2Client)
if err != nil {
logger.Error("Error querying ERC20 token details",
"l2_token", address.String(), "err", err)
"l2_token", withdrawal.L2Token.String(), "err", err)
token = &db.Token{
Address: address.String(),
Address: withdrawal.L2Token.String(),
}
}
if err := s.cfg.DB.AddL2Token(address.String(), token); err != nil {
if err := s.cfg.DB.AddL2Token(withdrawal.L2Token.String(), token); err != nil {
return err
}
s.tokenCache[address] = token
s.tokenCache[withdrawal.L2Token] = token
s.metrics.IncL2CachedTokensCount()
return nil
}
......@@ -489,16 +462,11 @@ func (s *Service) Start() error {
if s.cfg.ChainID == nil {
return errNoChainID
}
s.wg.Add(1)
go s.Loop(s.ctx)
go s.loop()
return nil
}
func (s *Service) Stop() {
s.cancel()
s.wg.Wait()
err := s.cfg.DB.Close()
if err != nil {
logger.Error("Error closing db", "err", err)
}
}
indexer/services/l2/query.go → indexer/services/query/erc20.go
View file @ 1636250a
package l2
package query
import (
"github.com/ethereum-optimism/optimism/indexer/db"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/ethclient"
)
func QueryERC20(address common.Address, client *ethclient.Client) (*db.Token, error) {
func NewERC20(address common.Address, client *ethclient.Client) (*db.Token, error) {
contract, err := bindings.NewERC20(address, client)
if err != nil {
return nil, err
......
indexer/services/query/headers.go 0 → 100644
View file @ 1636250a
package query
import (
"context"
"github.com/ethereum-optimism/optimism/op-service/backoff"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/ethclient"
)
// HeaderByNumberWithRetry retries getting headers.
func HeaderByNumberWithRetry(ctx context.Context, client *ethclient.Client) (*types.Header, error) {
var res *types.Header
err := backoff.DoCtx(ctx, 3, backoff.Exponential(), func() error {
var err error
res, err = client.HeaderByNumber(ctx, nil)
return err
})
return res, err
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment