Skip to content

N
nebula
Commit 2aa13672 authored by Hamdi Allam's avatar Hamdi Allam Committed by GitHub
Browse files
Options

feat(indexer) bridge fast sync (#8607) 

* bridge fast sync

* updates

* more conservative blocks limit
parent 54305b4e
Show whitespace changes
Inline Side-by-side
Showing with 144 additions and 79 deletions
indexer/processors/bridge.go
View file @ 2aa13672
......@@ -8,6 +8,7 @@ import (
"gorm.io/gorm"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum-optimism/optimism/indexer/bigint"
......@@ -18,7 +19,7 @@ import (
"github.com/ethereum-optimism/optimism/op-service/tasks"
)
var blocksLimit = 10_000
var blocksLimit = 500
type BridgeProcessor struct {
log log.Logger
......@@ -88,13 +89,14 @@ func NewBridgeProcessor(log log.Logger, db *database.DB, metrics bridge.Metricer
func (b *BridgeProcessor) Start() error {
b.log.Info("starting bridge processor...")
// start L1 worker
b.tasks.Go(func() error {
l1EtlUpdates := b.l1Etl.Notify()
for range l1EtlUpdates {
done := b.metrics.RecordL1Interval()
done(b.onL1Data())
b.log.Info("notified of traversed L1 state", "l1_etl_block_number", b.l1Etl.LatestHeader.Number)
if err := b.onL1Data(b.l1Etl.LatestHeader); err != nil {
b.log.Error("failed l1 bridge processing interval", "err", err)
}
}
b.log.Info("no more l1 etl updates. shutting down l1 task")
return nil
......@@ -103,8 +105,10 @@ func (b *BridgeProcessor) Start() error {
b.tasks.Go(func() error {
l2EtlUpdates := b.l2Etl.Notify()
for range l2EtlUpdates {
done := b.metrics.RecordL2Interval()
done(b.onL2Data())
b.log.Info("notified of traversed L2 state", "l2_etl_block_number", b.l2Etl.LatestHeader.Number)
if err := b.onL2Data(b.l2Etl.LatestHeader); err != nil {
b.log.Error("failed l2 bridge processing interval", "err", err)
}
}
b.log.Info("no more l2 etl updates. shutting down l2 task")
return nil
......@@ -121,23 +125,35 @@ func (b *BridgeProcessor) Close() error {
// onL1Data will index new bridge events for the unvisited L1 state. As new L1 bridge events
// are processed, bridge finalization events can be processed on L2 in this same window.
func (b *BridgeProcessor) onL1Data() error {
latestL1Header := b.l1Etl.LatestHeader
b.log.Info("notified of new L1 state", "l1_etl_block_number", latestL1Header.Number)
func (b *BridgeProcessor) onL1Data(latestL1Header *types.Header) (errs error) {
var errs error
if err := b.processInitiatedL1Events(); err != nil {
b.log.Error("failed to process initiated L1 events", "err", err)
errs = errors.Join(errs, err)
// Continue while unvisited state is available to process
for errs == nil {
done := b.metrics.RecordL1Interval()
lastL1Header := b.LastL1Header
lastFinalizedL2Header := b.LastFinalizedL2Header
// Initiated L1 Events
if b.LastL1Header == nil || b.LastL1Header.Timestamp < latestL1Header.Time {
if err := b.processInitiatedL1Events(latestL1Header); err != nil {
errs = errors.Join(errs, fmt.Errorf("failed processing initiated l1 events: %w", err))
}
}
// `LastFinalizedL2Header` and `LastL1Header` are mutated by the same routine and can
// safely be read without needing any sync primitives. Not every L1 block is indexed
// so check against a false interval on start.
if b.LastL1Header != nil && (b.LastFinalizedL2Header == nil || b.LastFinalizedL2Header.Timestamp < b.LastL1Header.Timestamp) {
if err := b.processFinalizedL2Events(); err != nil {
b.log.Error("failed to process finalized L2 events", "err", err)
errs = errors.Join(errs, err)
// Finalized L1 Events (on L2)
// - Not every L1 block is indexed so check against a false interval on start.
if b.LastL1Header != nil && (b.LastFinalizedL2Header == nil || b.LastFinalizedL2Header.Timestamp < latestL1Header.Time) {
if err := b.processFinalizedL2Events(latestL1Header); err != nil {
errs = errors.Join(errs, fmt.Errorf("failed processing finalized l2 events: %w", err))
}
}
done(errs)
// Break if there has been no change in processed events.
if lastL1Header == b.LastL1Header && lastFinalizedL2Header == b.LastFinalizedL2Header {
break
}
}
......@@ -146,24 +162,37 @@ func (b *BridgeProcessor) onL1Data() error {
// onL2Data will index new bridge events for the unvisited L2 state. As new L2 bridge events
// are processed, bridge finalization events can be processed on L1 in this same window.
func (b *BridgeProcessor) onL2Data() error {
if b.l2Etl.LatestHeader.Number.Cmp(bigint.Zero) == 0 {
func (b *BridgeProcessor) onL2Data(latestL2Header *types.Header) (errs error) {
if latestL2Header.Number.Cmp(bigint.Zero) == 0 {
return nil // skip genesis
}
b.log.Info("notified of new L2 state", "l2_etl_block_number", b.l2Etl.LatestHeader.Number)
var errs error
if err := b.processInitiatedL2Events(); err != nil {
b.log.Error("failed to process initiated L2 events", "err", err)
errs = errors.Join(errs, err)
// Continue while unvisited state is available to process
for errs == nil {
done := b.metrics.RecordL2Interval()
lastL2Header := b.LastL2Header
lastFinalizedL1Header := b.LastFinalizedL1Header
// Initiated L2 Events
if b.LastL2Header == nil || b.LastL2Header.Timestamp < latestL2Header.Time {
if err := b.processInitiatedL2Events(latestL2Header); err != nil {
errs = errors.Join(errs, fmt.Errorf("failed processing initiated l2 events: %w", err))
}
}
// Finalized L2 Events (on L1)
if b.LastFinalizedL1Header == nil || b.LastFinalizedL1Header.Timestamp < latestL2Header.Time {
if err := b.processFinalizedL1Events(latestL2Header); err != nil {
errs = errors.Join(errs, fmt.Errorf("failed processing finalized l1 events: %w", err))
}
}
done(errs)
// `LastFinalizedL1Header` and `LastL2Header` are mutated by the same routine and can
// safely be read without needing any sync primitives
if b.LastFinalizedL1Header == nil || b.LastFinalizedL1Header.Timestamp < b.LastL2Header.Timestamp {
if err := b.processFinalizedL1Events(); err != nil {
b.log.Error("failed to process finalized L1 events", "err", err)
errs = errors.Join(errs, err)
// Break if there has been no change in processed events.
if lastL2Header == b.LastL2Header && lastFinalizedL1Header == b.LastFinalizedL1Header {
break
}
}
......@@ -172,29 +201,29 @@ func (b *BridgeProcessor) onL2Data() error {
// Process Initiated Bridge Events
func (b *BridgeProcessor) processInitiatedL1Events() error {
func (b *BridgeProcessor) processInitiatedL1Events(latestL1Header *types.Header) error {
l1BridgeLog := b.log.New("bridge", "l1", "kind", "initiated")
lastL1BlockNumber := big.NewInt(int64(b.chainConfig.L1StartingHeight) - 1)
lastL1BlockNumber := big.NewInt(int64(b.chainConfig.L1StartingHeight - 1))
if b.LastL1Header != nil {
lastL1BlockNumber = b.LastL1Header.Number
}
// Latest unobserved L1 state bounded by `blockLimits` blocks. Since
// not every L1 block is indexed, we may have nothing to process.
latestL1HeaderScope := func(db *gorm.DB) *gorm.DB {
toL1HeaderScope := func(db *gorm.DB) *gorm.DB {
newQuery := db.Session(&gorm.Session{NewDB: true}) // fresh subquery
headers := newQuery.Model(database.L1BlockHeader{}).Where("number > ?", lastL1BlockNumber)
return db.Where("number = (?)", newQuery.Table("(?) as block_numbers", headers.Order("number ASC").Limit(blocksLimit)).Select("MAX(number)"))
headers := newQuery.Model(database.L1BlockHeader{}).Where("number > ? AND number <= ?", lastL1BlockNumber, latestL1Header.Number)
return db.Where("number = (?)", newQuery.Table("(?) AS block_numbers", headers.Order("number ASC").Limit(blocksLimit)).Select("MAX(number)"))
}
latestL1Header, err := b.db.Blocks.L1BlockHeaderWithScope(latestL1HeaderScope)
toL1Header, err := b.db.Blocks.L1BlockHeaderWithScope(toL1HeaderScope)
if err != nil {
return fmt.Errorf("failed to query new L1 state: %w", err)
} else if latestL1Header == nil {
} else if toL1Header == nil {
l1BridgeLog.Debug("no new L1 state found")
return nil
}
fromL1Height, toL1Height := new(big.Int).Add(lastL1BlockNumber, bigint.One), latestL1Header.Number
fromL1Height, toL1Height := new(big.Int).Add(lastL1BlockNumber, bigint.One), toL1Header.Number
if err := b.db.Transaction(func(tx *database.DB) error {
l1BedrockStartingHeight := big.NewInt(int64(b.chainConfig.L1BedrockStartingHeight))
if l1BedrockStartingHeight.Cmp(fromL1Height) > 0 { // OP Mainnet & OP Goerli Only.
......@@ -221,12 +250,12 @@ func (b *BridgeProcessor) processInitiatedL1Events() error {
return err
}
b.LastL1Header = latestL1Header
b.metrics.RecordL1LatestHeight(latestL1Header.Number)
b.LastL1Header = toL1Header
b.metrics.RecordL1LatestHeight(toL1Header.Number)
return nil
}
func (b *BridgeProcessor) processInitiatedL2Events() error {
func (b *BridgeProcessor) processInitiatedL2Events(latestL2Header *types.Header) error {
l2BridgeLog := b.log.New("bridge", "l2", "kind", "initiated")
lastL2BlockNumber := bigint.Zero
if b.LastL2Header != nil {
......@@ -235,19 +264,19 @@ func (b *BridgeProcessor) processInitiatedL2Events() error {
// Latest unobserved L2 state bounded by `blockLimits` blocks.
// Since every L2 block is indexed, we always expect new state.
latestL2HeaderScope := func(db *gorm.DB) *gorm.DB {
toL2HeaderScope := func(db *gorm.DB) *gorm.DB {
newQuery := db.Session(&gorm.Session{NewDB: true}) // fresh subquery
headers := newQuery.Model(database.L2BlockHeader{}).Where("number > ?", lastL2BlockNumber)
return db.Where("number = (?)", newQuery.Table("(?) as block_numbers", headers.Order("number ASC").Limit(blocksLimit)).Select("MAX(number)"))
headers := newQuery.Model(database.L2BlockHeader{}).Where("number > ? AND number <= ?", lastL2BlockNumber, latestL2Header.Number)
return db.Where("number = (?)", newQuery.Table("(?) AS block_numbers", headers.Order("number ASC").Limit(blocksLimit)).Select("MAX(number)"))
}
latestL2Header, err := b.db.Blocks.L2BlockHeaderWithScope(latestL2HeaderScope)
toL2Header, err := b.db.Blocks.L2BlockHeaderWithScope(toL2HeaderScope)
if err != nil {
return fmt.Errorf("failed to query new L2 state: %w", err)
} else if latestL2Header == nil {
} else if toL2Header == nil {
return fmt.Errorf("no new L2 state found")
}
fromL2Height, toL2Height := new(big.Int).Add(lastL2BlockNumber, bigint.One), latestL2Header.Number
fromL2Height, toL2Height := new(big.Int).Add(lastL2BlockNumber, bigint.One), toL2Header.Number
if err := b.db.Transaction(func(tx *database.DB) error {
l2BedrockStartingHeight := big.NewInt(int64(b.chainConfig.L2BedrockStartingHeight))
if l2BedrockStartingHeight.Cmp(fromL2Height) > 0 { // OP Mainnet & OP Goerli Only
......@@ -274,14 +303,14 @@ func (b *BridgeProcessor) processInitiatedL2Events() error {
return err
}
b.LastL2Header = latestL2Header
b.metrics.RecordL2LatestHeight(latestL2Header.Number)
b.LastL2Header = toL2Header
b.metrics.RecordL2LatestHeight(toL2Header.Number)
return nil
}
// Process Finalized Bridge Events
func (b *BridgeProcessor) processFinalizedL1Events() error {
func (b *BridgeProcessor) processFinalizedL1Events(latestL2Header *types.Header) error {
l1BridgeLog := b.log.New("bridge", "l1", "kind", "finalization")
lastFinalizedL1BlockNumber := big.NewInt(int64(b.chainConfig.L1StartingHeight) - 1)
if b.LastFinalizedL1Header != nil {
......@@ -290,20 +319,20 @@ func (b *BridgeProcessor) processFinalizedL1Events() error {
// Latest unfinalized L1 state bounded by `blockLimit` blocks that have had L2 bridge events
// indexed. Since L1 data is indexed independently, there may not be new L1 state to finalize
latestL1HeaderScope := func(db *gorm.DB) *gorm.DB {
toL1HeaderScope := func(db *gorm.DB) *gorm.DB {
newQuery := db.Session(&gorm.Session{NewDB: true}) // fresh subquery
headers := newQuery.Model(database.L1BlockHeader{}).Where("number > ? AND timestamp <= ?", lastFinalizedL1BlockNumber, b.LastL2Header.Timestamp)
return db.Where("number = (?)", newQuery.Table("(?) as block_numbers", headers.Order("number ASC").Limit(blocksLimit)).Select("MAX(number)"))
headers := newQuery.Model(database.L1BlockHeader{}).Where("number > ? AND timestamp <= ?", lastFinalizedL1BlockNumber, latestL2Header.Time)
return db.Where("number = (?)", newQuery.Table("(?) AS block_numbers", headers.Order("number ASC").Limit(blocksLimit)).Select("MAX(number)"))
}
latestL1Header, err := b.db.Blocks.L1BlockHeaderWithScope(latestL1HeaderScope)
toL1Header, err := b.db.Blocks.L1BlockHeaderWithScope(toL1HeaderScope)
if err != nil {
return fmt.Errorf("failed to query for latest unfinalized L1 state: %w", err)
} else if latestL1Header == nil {
} else if toL1Header == nil {
l1BridgeLog.Debug("no new l1 state to finalize", "last_finalized_block_number", lastFinalizedL1BlockNumber)
return nil
}
fromL1Height, toL1Height := new(big.Int).Add(lastFinalizedL1BlockNumber, bigint.One), latestL1Header.Number
fromL1Height, toL1Height := new(big.Int).Add(lastFinalizedL1BlockNumber, bigint.One), toL1Header.Number
if err := b.db.Transaction(func(tx *database.DB) error {
l1BedrockStartingHeight := big.NewInt(int64(b.chainConfig.L1BedrockStartingHeight))
if l1BedrockStartingHeight.Cmp(fromL1Height) > 0 {
......@@ -330,12 +359,12 @@ func (b *BridgeProcessor) processFinalizedL1Events() error {
return err
}
b.LastFinalizedL1Header = latestL1Header
b.metrics.RecordL1LatestFinalizedHeight(latestL1Header.Number)
b.LastFinalizedL1Header = toL1Header
b.metrics.RecordL1LatestFinalizedHeight(toL1Header.Number)
return nil
}
func (b *BridgeProcessor) processFinalizedL2Events() error {
func (b *BridgeProcessor) processFinalizedL2Events(latestL1Header *types.Header) error {
l2BridgeLog := b.log.New("bridge", "l2", "kind", "finalization")
lastFinalizedL2BlockNumber := bigint.Zero
if b.LastFinalizedL2Header != nil {
......@@ -344,20 +373,20 @@ func (b *BridgeProcessor) processFinalizedL2Events() error {
// Latest unfinalized L2 state bounded by `blockLimit` blocks that have had L1 bridge events
// indexed. Since L2 data is indexed independently, there may not be new L2 state to finalize
latestL2HeaderScope := func(db *gorm.DB) *gorm.DB {
toL2HeaderScope := func(db *gorm.DB) *gorm.DB {
newQuery := db.Session(&gorm.Session{NewDB: true}) // fresh subquery
headers := newQuery.Model(database.L2BlockHeader{}).Where("number > ? AND timestamp <= ?", lastFinalizedL2BlockNumber, b.LastL1Header.Timestamp)
return db.Where("number = (?)", newQuery.Table("(?) as block_numbers", headers.Order("number ASC").Limit(blocksLimit)).Select("MAX(number)"))
headers := newQuery.Model(database.L2BlockHeader{}).Where("number > ? AND timestamp <= ?", lastFinalizedL2BlockNumber, latestL1Header.Time)
return db.Where("number = (?)", newQuery.Table("(?) AS block_numbers", headers.Order("number ASC").Limit(blocksLimit)).Select("MAX(number)"))
}
latestL2Header, err := b.db.Blocks.L2BlockHeaderWithScope(latestL2HeaderScope)
toL2Header, err := b.db.Blocks.L2BlockHeaderWithScope(toL2HeaderScope)
if err != nil {
return fmt.Errorf("failed to query for latest unfinalized L2 state: %w", err)
} else if latestL2Header == nil {
} else if toL2Header == nil {
l2BridgeLog.Debug("no new l2 state to finalize", "last_finalized_block_number", lastFinalizedL2BlockNumber)
return nil
}
fromL2Height, toL2Height := new(big.Int).Add(lastFinalizedL2BlockNumber, bigint.One), latestL2Header.Number
fromL2Height, toL2Height := new(big.Int).Add(lastFinalizedL2BlockNumber, bigint.One), toL2Header.Number
if err := b.db.Transaction(func(tx *database.DB) error {
l2BedrockStartingHeight := big.NewInt(int64(b.chainConfig.L2BedrockStartingHeight))
if l2BedrockStartingHeight.Cmp(fromL2Height) > 0 {
......@@ -384,7 +413,7 @@ func (b *BridgeProcessor) processFinalizedL2Events() error {
return err
}
b.LastFinalizedL2Header = latestL2Header
b.metrics.RecordL2LatestFinalizedHeight(latestL2Header.Number)
b.LastFinalizedL2Header = toL2Header
b.metrics.RecordL2LatestFinalizedHeight(toL2Header.Number)
return nil
}
indexer/processors/bridge/legacy_bridge_processor.go
View file @ 2aa13672
......@@ -178,12 +178,12 @@ func LegacyL2ProcessInitiatedBridgeEvents(log log.Logger, db *database.DB, metri
// require the entry but we'll store it anyways as a large % of these withdrawals are not relayed
// pre-bedrock.
v1MessageHash, err := legacyBridgeMessageV1MessageHash(&sentMessage.BridgeMessage)
v1MessageHash, err := LegacyBridgeMessageV1MessageHash(&sentMessage.BridgeMessage)
if err != nil {
return fmt.Errorf("failed to compute versioned message hash: %w", err)
}
withdrawalHash, err := legacyBridgeMessageWithdrawalHash(preset, &sentMessage.BridgeMessage)
withdrawalHash, err := LegacyBridgeMessageWithdrawalHash(preset, &sentMessage.BridgeMessage)
if err != nil {
return fmt.Errorf("failed to construct migrated withdrawal hash: %w", err)
}
......@@ -373,7 +373,7 @@ func LegacyL2ProcessFinalizedBridgeEvents(log log.Logger, db *database.DB, metri
// Utils
func legacyBridgeMessageWithdrawalHash(preset int, msg *database.BridgeMessage) (common.Hash, error) {
func LegacyBridgeMessageWithdrawalHash(preset int, msg *database.BridgeMessage) (common.Hash, error) {
l1Cdm := config.Presets[preset].ChainConfig.L1Contracts.L1CrossDomainMessengerProxy
legacyWithdrawal := crossdomain.NewLegacyWithdrawal(predeploys.L2CrossDomainMessengerAddr, msg.Tx.ToAddress, msg.Tx.FromAddress, msg.Tx.Data, msg.Nonce)
migratedWithdrawal, err := crossdomain.MigrateWithdrawal(legacyWithdrawal, &l1Cdm, big.NewInt(int64(preset)))
......@@ -384,7 +384,7 @@ func legacyBridgeMessageWithdrawalHash(preset int, msg *database.BridgeMessage)
return migratedWithdrawal.Hash()
}
func legacyBridgeMessageV1MessageHash(msg *database.BridgeMessage) (common.Hash, error) {
func LegacyBridgeMessageV1MessageHash(msg *database.BridgeMessage) (common.Hash, error) {
legacyWithdrawal := crossdomain.NewLegacyWithdrawal(predeploys.L2CrossDomainMessengerAddr, msg.Tx.ToAddress, msg.Tx.FromAddress, msg.Tx.Data, msg.Nonce)
value, err := legacyWithdrawal.Value()
if err != nil {
......
indexer/processors/bridge/legacy_bridge_processor_test.go
View file @ 2aa13672
package bridge
import (
"math/big"
"testing"
"github.com/ethereum-optimism/optimism/indexer/bigint"
......@@ -37,13 +38,48 @@ func TestLegacyWithdrawalAndMessageHash(t *testing.T) {
Tx: database.Transaction{FromAddress: sentMessage.Sender, ToAddress: sentMessage.Target, Amount: value, Data: sentMessage.Message},
}
hash, err := legacyBridgeMessageWithdrawalHash(420, &msg)
hash, err := LegacyBridgeMessageWithdrawalHash(420, &msg)
require.NoError(t, err)
require.Equal(t, expectedWithdrawalHash, hash)
// Ensure the relayed message hash (v1) matches
expectedMessageHash := common.HexToHash("0xcb16ecc1967f5d7aed909349a4351d28fbb396429ef7faf1c9d2a670e3ca906f")
v1MessageHash, err := legacyBridgeMessageV1MessageHash(&msg)
v1MessageHash, err := LegacyBridgeMessageV1MessageHash(&msg)
require.NoError(t, err)
require.Equal(t, expectedMessageHash, v1MessageHash)
// OP Mainnet hashes to also check for
// - since the message hash doesn't depend on the preset, we only need to check for the withdrawal hash
expectedWithdrawalHash = common.HexToHash("0x9c0bc28a77328a405f21d51a32d32f038ebf7ce70e377ca48b2cd194ec024f15")
msg = database.BridgeMessage{
Nonce: big.NewInt(100180),
GasLimit: bigint.Zero,
Tx: database.Transaction{
FromAddress: common.HexToAddress("0x4200000000000000000000000000000000000010"),
ToAddress: common.HexToAddress("0x99c9fc46f92e8a1c0dec1b1747d010903e884be1"),
Amount: bigint.Zero,
Data: common.FromHex("0x1532ec34000000000000000000000000094a9009fe93a85658e4b49604fd8177620f8cd8000000000000000000000000094a9009fe93a85658e4b49604fd8177620f8cd8000000000000000000000000000000000000000000000000013abb2a2774ab0000000000000000000000000000000000000000000000000000000000000000800000000000000000000000000000000000000000000000000000000000000000"),
},
}
hash, err = LegacyBridgeMessageWithdrawalHash(10, &msg)
require.NoError(t, err)
require.Equal(t, expectedWithdrawalHash, hash)
expectedWithdrawalHash = common.HexToHash("0xeb1dd5ead967ad6860d64407413f86e50330ab123ca9adf2768145524c3f5323")
msg = database.BridgeMessage{
Nonce: big.NewInt(100618),
GasLimit: bigint.Zero,
Tx: database.Transaction{
FromAddress: common.HexToAddress("0x4200000000000000000000000000000000000010"),
ToAddress: common.HexToAddress("0x99c9fc46f92e8a1c0dec1b1747d010903e884be1"),
Amount: bigint.Zero,
Data: common.FromHex("0xa9f9e67500000000000000000000000028e1de268616a6ba0de59717ac5547589e6bb1180000000000000000000000003ef241d9ae02f2253d8a1bf0b35d68eab9925b400000000000000000000000003e579180cf01f0e2abf6ff4d566b7891fbf9b8680000000000000000000000003e579180cf01f0e2abf6ff4d566b7891fbf9b868000000000000000000000000000000000000000000000000000000174876e80000000000000000000000000000000000000000000000000000000000000000c00000000000000000000000000000000000000000000000000000000000000000"),
},
}
hash, err = LegacyBridgeMessageWithdrawalHash(10, &msg)
require.NoError(t, err)
require.Equal(t, expectedWithdrawalHash, hash)
}
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