Skip to content

N
nebula
Commit 3ec079da authored by OptimismBot's avatar OptimismBot Committed by GitHub
Browse files
Options

Merge pull request #7161 from ethereum-optimism/indexer.bridge.interval 

fix(indexer): run bridge processor on startup & split large unobserved epochs
parents 67e74078 0637b004
Hide whitespace changes
Inline Side-by-side
Showing with 86 additions and 59 deletions
indexer/processors/bridge.go
View file @ 3ec079da
......@@ -21,8 +21,6 @@ type BridgeProcessor struct {
l1Etl *etl.L1ETL
chainConfig config.ChainConfig
// NOTE: We'll need this processor to handle for reorgs events.
LatestL1Header *types.Header
LatestL2Header *types.Header
}
......@@ -52,7 +50,7 @@ func NewBridgeProcessor(log log.Logger, db *database.DB, l1Etl *etl.L1ETL, chain
l2Height = latestL2Header.Number
l2Header = latestL2Header.RLPHeader.Header()
}
log.Info("detected latest indexed state", "l1_block_number", l1Height, "l2_block_number", l2Height)
log.Info("detected latest indexed bridge state", "l1_block_number", l1Height, "l2_block_number", l2Height)
}
return &BridgeProcessor{log, db, l1Etl, chainConfig, l1Header, l2Header}, nil
......@@ -70,6 +68,9 @@ func (b *BridgeProcessor) Start(ctx context.Context) error {
// serves as this shared marker.
l1EtlUpdates := b.l1Etl.Notify()
startup := make(chan interface{}, 1)
startup <- nil
b.log.Info("starting bridge processor...")
for {
select {
......@@ -77,81 +78,107 @@ func (b *BridgeProcessor) Start(ctx context.Context) error {
b.log.Info("stopping bridge processor")
return nil
// Fire off independently on startup to check for any
// new data or if we've indexed new L1 data.
case <-startup:
case <-l1EtlUpdates:
latestEpoch, err := b.db.Blocks.LatestEpoch()
if err != nil {
return err
} else if latestEpoch == nil {
if b.LatestL1Header != nil || b.LatestL2Header != nil {
// Once we have some indexed state `latestEpoch` can never return nil
b.log.Error("bridge events indexed, but no indexed epoch returned", "latest_bridge_l1_block_number", b.LatestL1Header.Number)
return errors.New("bridge events indexed, but no indexed epoch returned")
}
}
b.log.Warn("no indexed epochs available. waiting...")
continue
latestEpoch, err := b.db.Blocks.LatestEpoch()
if err != nil {
return err
} else if latestEpoch == nil {
if b.LatestL1Header != nil || b.LatestL2Header != nil {
// Once we have some indexed state `latestEpoch` can never return nil
b.log.Error("bridge events indexed, but no indexed epoch returned", "latest_bridge_l1_block_number", b.LatestL1Header.Number)
return errors.New("bridge events indexed, but no indexed epoch returned")
}
// Integrity Checks
b.log.Warn("no indexed epochs available. waiting...")
continue
}
if b.LatestL1Header != nil && latestEpoch.L1BlockHeader.Hash == b.LatestL1Header.Hash() {
b.log.Warn("all available epochs indexed", "latest_bridge_l1_block_number", b.LatestL1Header.Number)
continue
}
if b.LatestL1Header != nil && latestEpoch.L1BlockHeader.Number.Cmp(b.LatestL1Header.Number) <= 0 {
b.log.Error("non-increasing l1 block height observed", "latest_bridge_l1_block_number", b.LatestL1Header.Number, "latest_epoch_number", latestEpoch.L1BlockHeader.Number)
return errors.New("non-increasing l1 block heght observed")
}
if b.LatestL2Header != nil && latestEpoch.L2BlockHeader.Number.Cmp(b.LatestL2Header.Number) <= 0 {
b.log.Error("non-increasing l2 block height observed", "latest_bridge_l2_block_number", b.LatestL2Header.Number, "latest_epoch_number", latestEpoch.L2BlockHeader.Number)
return errors.New("non-increasing l2 block heght observed")
}
// Integrity Checks
// Process Bridge Events
if b.LatestL1Header != nil && latestEpoch.L1BlockHeader.Hash == b.LatestL1Header.Hash() {
b.log.Warn("all available epochs indexed", "latest_bridge_l1_block_number", b.LatestL1Header.Number)
continue
}
if b.LatestL1Header != nil && latestEpoch.L1BlockHeader.Number.Cmp(b.LatestL1Header.Number) <= 0 {
b.log.Error("decreasing l1 block height observed", "latest_bridge_l1_block_number", b.LatestL1Header.Number, "latest_epoch_number", latestEpoch.L1BlockHeader.Number)
return errors.New("decreasing l1 block heght observed")
}
if b.LatestL2Header != nil && latestEpoch.L2BlockHeader.Number.Cmp(b.LatestL2Header.Number) <= 0 {
b.log.Error("decreasing l2 block height observed", "latest_bridge_l2_block_number", b.LatestL2Header.Number, "latest_epoch_number", latestEpoch.L2BlockHeader.Number)
return errors.New("decreasing l2 block heght observed")
}
toL1Height, toL2Height := latestEpoch.L1BlockHeader.Number, latestEpoch.L2BlockHeader.Number
fromL1Height, fromL2Height := bigint.Zero, bigint.Zero
if b.LatestL1Header != nil {
fromL1Height = new(big.Int).Add(b.LatestL1Header.Number, bigint.One)
}
if b.LatestL2Header != nil {
fromL2Height = new(big.Int).Add(b.LatestL2Header.Number, bigint.One)
}
// Process Bridge Events
batchLog := b.log.New("epoch_start_number", fromL1Height, "epoch_end_number", toL1Height)
batchLog.Info("scanning for new bridge events")
err = b.db.Transaction(func(tx *database.DB) error {
l1BridgeLog := b.log.New("from_l1_block_number", fromL1Height, "to_l1_block_number", toL1Height)
l2BridgeLog := b.log.New("from_l2_block_number", fromL2Height, "to_l2_block_number", toL2Height)
toL1Height, toL2Height := latestEpoch.L1BlockHeader.Number, latestEpoch.L2BlockHeader.Number
fromL1Height, fromL2Height := big.NewInt(int64(b.chainConfig.L1StartingHeight)), bigint.Zero
if b.LatestL1Header != nil {
fromL1Height = new(big.Int).Add(b.LatestL1Header.Number, bigint.One)
}
if b.LatestL2Header != nil {
fromL2Height = new(big.Int).Add(b.LatestL2Header.Number, bigint.One)
}
// First, find all possible initiated bridge events
if err := bridge.L1ProcessInitiatedBridgeEvents(l1BridgeLog, tx, b.chainConfig, fromL1Height, toL1Height); err != nil {
batchLog := b.log.New("epoch_start_number", fromL1Height, "epoch_end_number", toL1Height)
batchLog.Info("unobserved epochs")
err = b.db.Transaction(func(tx *database.DB) error {
l1BridgeLog := b.log.New("bridge", "l1")
l2BridgeLog := b.log.New("bridge", "l2")
// In the event where we have a large number of un-observed blocks, group the block range
// on the order of 10k blocks at a time. If this turns out to be a bottleneck, we can
// parallelize these operations for significant improvements as well
l1BlockGroups := bigint.Grouped(fromL1Height, toL1Height, 10_000)
l2BlockGroups := bigint.Grouped(fromL2Height, toL2Height, 10_000)
// First, find all possible initiated bridge events
for _, group := range l1BlockGroups {
log := l1BridgeLog.New("from_block_number", group.Start, "to_block_number", group.End)
log.Info("scanning for initiated bridge events")
if err := bridge.L1ProcessInitiatedBridgeEvents(log, tx, b.chainConfig, group.Start, group.End); err != nil {
return err
}
if err := bridge.L2ProcessInitiatedBridgeEvents(l2BridgeLog, tx, fromL2Height, toL2Height); err != nil {
}
for _, group := range l2BlockGroups {
log := l2BridgeLog.New("from_block_number", group.Start, "to_block_number", group.End)
log.Info("scanning for initiated bridge events")
if err := bridge.L2ProcessInitiatedBridgeEvents(log, tx, group.Start, group.End); err != nil {
return err
}
}
// Now that all initiated events have been indexed, it is ensured that all finalization can find their counterpart.
if err := bridge.L1ProcessFinalizedBridgeEvents(l1BridgeLog, tx, b.chainConfig, fromL1Height, toL1Height); err != nil {
// Now all finalization events can find their counterpart.
for _, group := range l1BlockGroups {
log := l1BridgeLog.New("from_block_number", group.Start, "to_block_number", group.End)
log.Info("scanning for finalized bridge events")
if err := bridge.L1ProcessFinalizedBridgeEvents(log, tx, b.chainConfig, group.Start, group.End); err != nil {
return err
}
if err := bridge.L2ProcessFinalizedBridgeEvents(l2BridgeLog, tx, fromL2Height, toL2Height); err != nil {
}
for _, group := range l2BlockGroups {
log := l2BridgeLog.New("from_block_number", group.Start, "to_block_number", group.End)
log.Info("scanning for finalized bridge events")
if err := bridge.L2ProcessFinalizedBridgeEvents(log, tx, group.Start, group.End); err != nil {
return err
}
// a-ok
return nil
})
if err != nil {
// Try again on a subsequent interval
batchLog.Error("unable to index new bridge events", "err", err)
} else {
batchLog.Info("done indexing bridge events", "latest_l1_block_number", toL1Height, "latest_l2_block_number", toL2Height)
b.LatestL1Header = latestEpoch.L1BlockHeader.RLPHeader.Header()
b.LatestL2Header = latestEpoch.L2BlockHeader.RLPHeader.Header()
}
// a-ok
return nil
})
if err != nil {
// Try again on a subsequent interval
batchLog.Error("failed to index bridge events", "err", err)
} else {
batchLog.Info("indexed bridge events", "latest_l1_block_number", toL1Height, "latest_l2_block_number", toL2Height)
b.LatestL1Header = latestEpoch.L1BlockHeader.RLPHeader.Header()
b.LatestL2Header = latestEpoch.L2BlockHeader.RLPHeader.Header()
}
}
}
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