Commit 9226e8ff authored by Hamdi Allam's avatar Hamdi Allam Committed by GitHub

Merge pull request #6250 from ethereum-optimism/indexer.bridge.events

feat: bi-directionally index StandardBridge contract events
parents 49014a29 134e71aa
package api
import (
"math/big"
"net/http"
"net/http/httptest"
"testing"
......@@ -25,7 +26,7 @@ func (mbv *MockBridgeView) DepositsByAddress(address common.Address) ([]*databas
{
Deposit: database.Deposit{
GUID: uuid.MustParse(guid1),
InitiatedL1EventGUID: guid2,
InitiatedL1EventGUID: uuid.MustParse(guid2),
Tx: database.Transaction{},
TokenPair: database.TokenPair{},
},
......@@ -34,13 +35,28 @@ func (mbv *MockBridgeView) DepositsByAddress(address common.Address) ([]*databas
}, nil
}
// DepositsByAddress mocks returning deposits by an address
func (mbv *MockBridgeView) DepositByMessageNonce(nonce *big.Int) (*database.Deposit, error) {
return &database.Deposit{
GUID: uuid.MustParse(guid1),
InitiatedL1EventGUID: uuid.MustParse(guid2),
Tx: database.Transaction{},
TokenPair: database.TokenPair{},
}, nil
}
// LatestDepositMessageNonce mocks returning the latest cross domain message nonce for a deposit
func (mbv *MockBridgeView) LatestDepositMessageNonce() (*big.Int, error) {
return big.NewInt(0), nil
}
// WithdrawalsByAddress mocks returning withdrawals by an address
func (mbv *MockBridgeView) WithdrawalsByAddress(address common.Address) ([]*database.WithdrawalWithTransactionHashes, error) {
return []*database.WithdrawalWithTransactionHashes{
{
Withdrawal: database.Withdrawal{
GUID: uuid.MustParse(guid2),
InitiatedL2EventGUID: guid1,
InitiatedL2EventGUID: uuid.MustParse(guid1),
WithdrawalHash: common.HexToHash("0x456"),
Tx: database.Transaction{},
TokenPair: database.TokenPair{},
......@@ -50,6 +66,33 @@ func (mbv *MockBridgeView) WithdrawalsByAddress(address common.Address) ([]*data
}, nil
}
// WithdrawalsByMessageNonce mocks returning withdrawals by a withdrawal hash
func (mbv *MockBridgeView) WithdrawalByMessageNonce(nonce *big.Int) (*database.Withdrawal, error) {
return &database.Withdrawal{
GUID: uuid.MustParse(guid2),
InitiatedL2EventGUID: uuid.MustParse(guid1),
WithdrawalHash: common.HexToHash("0x456"),
Tx: database.Transaction{},
TokenPair: database.TokenPair{},
}, nil
}
// WithdrawalsByHash mocks returning withdrawals by a withdrawal hash
func (mbv *MockBridgeView) WithdrawalByHash(address common.Hash) (*database.Withdrawal, error) {
return &database.Withdrawal{
GUID: uuid.MustParse(guid2),
InitiatedL2EventGUID: uuid.MustParse(guid1),
WithdrawalHash: common.HexToHash("0x456"),
Tx: database.Transaction{},
TokenPair: database.TokenPair{},
}, nil
}
// LatestWithdrawalMessageNonce mocks returning the latest cross domain message nonce for a withdrawal
func (mbv *MockBridgeView) LatestWithdrawalMessageNonce() (*big.Int, error) {
return big.NewInt(0), nil
}
func TestHealthz(t *testing.T) {
api := NewApi(&MockBridgeView{})
request, err := http.NewRequest("GET", "/healthz", nil)
......
......@@ -5,6 +5,8 @@ import (
"errors"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/google/uuid"
"gorm.io/gorm"
......@@ -21,6 +23,15 @@ type BlockHeader struct {
Timestamp uint64
}
func BlockHeaderFromGethHeader(header *types.Header) BlockHeader {
return BlockHeader{
Hash: header.Hash(),
ParentHash: header.ParentHash,
Number: U256{Int: header.Number},
Timestamp: header.Time,
}
}
type L1BlockHeader struct {
BlockHeader
}
......@@ -41,8 +52,9 @@ type LegacyStateBatch struct {
}
type OutputProposal struct {
OutputRoot common.Hash `gorm:"primaryKey;serializer:json"`
L2BlockNumber U256
OutputRoot common.Hash `gorm:"primaryKey;serializer:json"`
L2BlockNumber U256
L1ContractEventGUID uuid.UUID
}
......
......@@ -2,6 +2,8 @@ package database
import (
"errors"
"fmt"
"math/big"
"gorm.io/gorm"
......@@ -30,7 +32,17 @@ type TokenPair struct {
type Deposit struct {
GUID uuid.UUID `gorm:"primaryKey"`
InitiatedL1EventGUID string
InitiatedL1EventGUID uuid.UUID
// Since we're only currently indexing a single StandardBridge,
// the message nonce serves as a unique identifier for this
// deposit. Once this generalizes to more than 1 deployed
// bridge, we need to include the `CrossDomainMessenger` address
// such that the (messenger_addr, nonce) is the unique identifier
// for a bridge msg
SentMessageNonce U256
FinalizedL2EventGUID *uuid.UUID
Tx Transaction `gorm:"embedded"`
TokenPair TokenPair `gorm:"embedded"`
......@@ -43,11 +55,19 @@ type DepositWithTransactionHash struct {
type Withdrawal struct {
GUID uuid.UUID `gorm:"primaryKey"`
InitiatedL2EventGUID string
InitiatedL2EventGUID uuid.UUID
// Since we're only currently indexing a single StandardBridge,
// the message nonce serves as a unique identifier for this
// withdrawal. Once this generalizes to more than 1 deployed
// bridge, we need to include the `CrossDomainMessenger` address
// such that the (messenger_addr, nonce) is the unique identifier
// for a bridge msg
SentMessageNonce U256
WithdrawalHash common.Hash `gorm:"serializer:json"`
ProvenL1EventGUID *string
FinalizedL1EventGUID *string
ProvenL1EventGUID *uuid.UUID
FinalizedL1EventGUID *uuid.UUID
Tx Transaction `gorm:"embedded"`
TokenPair TokenPair `gorm:"embedded"`
......@@ -63,16 +83,24 @@ type WithdrawalWithTransactionHashes struct {
type BridgeView interface {
DepositsByAddress(address common.Address) ([]*DepositWithTransactionHash, error)
DepositByMessageNonce(*big.Int) (*Deposit, error)
LatestDepositMessageNonce() (*big.Int, error)
WithdrawalsByAddress(address common.Address) ([]*WithdrawalWithTransactionHashes, error)
WithdrawalByMessageNonce(*big.Int) (*Withdrawal, error)
WithdrawalByHash(common.Hash) (*Withdrawal, error)
LatestWithdrawalMessageNonce() (*big.Int, error)
}
type BridgeDB interface {
BridgeView
StoreDeposits([]*Deposit) error
MarkFinalizedDepositEvent(uuid.UUID, uuid.UUID) error
StoreWithdrawals([]*Withdrawal) error
MarkProvenWithdrawalEvent(string, string) error
MarkFinalizedWithdrawalEvent(string, string) error
MarkProvenWithdrawalEvent(uuid.UUID, uuid.UUID) error
MarkFinalizedWithdrawalEvent(uuid.UUID, uuid.UUID) error
}
/**
......@@ -114,6 +142,46 @@ func (db *bridgeDB) DepositsByAddress(address common.Address) ([]*DepositWithTra
return deposits, nil
}
func (db *bridgeDB) DepositByMessageNonce(nonce *big.Int) (*Deposit, error) {
var deposit Deposit
result := db.gorm.First(&deposit, "sent_message_nonce = ?", U256{Int: nonce})
if result.Error != nil {
if errors.Is(result.Error, gorm.ErrRecordNotFound) {
return nil, nil
}
return nil, result.Error
}
return &deposit, nil
}
func (db *bridgeDB) LatestDepositMessageNonce() (*big.Int, error) {
var deposit Deposit
result := db.gorm.Order("sent_message_nonce DESC").Take(&deposit)
if result.Error != nil {
if errors.Is(result.Error, gorm.ErrRecordNotFound) {
return nil, nil
}
return nil, result.Error
}
return deposit.SentMessageNonce.Int, nil
}
func (db *bridgeDB) MarkFinalizedDepositEvent(guid, finalizationEventGUID uuid.UUID) error {
var deposit Deposit
result := db.gorm.First(&deposit, "guid = ?", guid)
if result.Error != nil {
return result.Error
}
deposit.FinalizedL2EventGUID = &finalizationEventGUID
result = db.gorm.Save(&deposit)
return result.Error
}
// Withdrawals
func (db *bridgeDB) StoreWithdrawals(withdrawals []*Withdrawal) error {
......@@ -121,7 +189,7 @@ func (db *bridgeDB) StoreWithdrawals(withdrawals []*Withdrawal) error {
return result.Error
}
func (db *bridgeDB) MarkProvenWithdrawalEvent(guid, provenL1EventGuid string) error {
func (db *bridgeDB) MarkProvenWithdrawalEvent(guid, provenL1EventGuid uuid.UUID) error {
var withdrawal Withdrawal
result := db.gorm.First(&withdrawal, "guid = ?", guid)
if result.Error != nil {
......@@ -133,13 +201,17 @@ func (db *bridgeDB) MarkProvenWithdrawalEvent(guid, provenL1EventGuid string) er
return result.Error
}
func (db *bridgeDB) MarkFinalizedWithdrawalEvent(guid, finalizedL1EventGuid string) error {
func (db *bridgeDB) MarkFinalizedWithdrawalEvent(guid, finalizedL1EventGuid uuid.UUID) error {
var withdrawal Withdrawal
result := db.gorm.First(&withdrawal, "guid = ?", guid)
if result.Error != nil {
return result.Error
}
if withdrawal.ProvenL1EventGUID == nil {
return fmt.Errorf("withdrawal %s marked finalized prior to being proven", guid)
}
withdrawal.FinalizedL1EventGUID = &finalizedL1EventGuid
result = db.gorm.Save(&withdrawal)
return result.Error
......@@ -167,3 +239,45 @@ func (db *bridgeDB) WithdrawalsByAddress(address common.Address) ([]*WithdrawalW
return withdrawals, nil
}
func (db *bridgeDB) WithdrawalByMessageNonce(nonce *big.Int) (*Withdrawal, error) {
var withdrawal Withdrawal
result := db.gorm.First(&withdrawal, "sent_message_nonce = ?", U256{Int: nonce})
if result.Error != nil {
if errors.Is(result.Error, gorm.ErrRecordNotFound) {
return nil, nil
}
return nil, result.Error
}
return &withdrawal, nil
}
func (db *bridgeDB) WithdrawalByHash(hash common.Hash) (*Withdrawal, error) {
var withdrawal Withdrawal
result := db.gorm.First(&withdrawal, "withdrawal_hash = ?", hash.String())
if result.Error != nil {
if errors.Is(result.Error, gorm.ErrRecordNotFound) {
return nil, nil
}
return nil, result.Error
}
return &withdrawal, nil
}
func (db *bridgeDB) LatestWithdrawalMessageNonce() (*big.Int, error) {
var withdrawal Withdrawal
result := db.gorm.Order("sent_message_nonce DESC").Take(&withdrawal)
if result.Error != nil {
if errors.Is(result.Error, gorm.ErrRecordNotFound) {
return nil, nil
}
return nil, result.Error
}
return withdrawal.SentMessageNonce.Int, nil
}
......@@ -4,6 +4,7 @@ import (
"gorm.io/gorm"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/google/uuid"
)
......@@ -22,6 +23,20 @@ type ContractEvent struct {
Timestamp uint64
}
func ContractEventFromGethLog(log *types.Log, timestamp uint64) ContractEvent {
return ContractEvent{
GUID: uuid.New(),
BlockHash: log.BlockHash,
TransactionHash: log.TxHash,
EventSignature: log.Topics[0],
LogIndex: uint64(log.Index),
Timestamp: timestamp,
}
}
type L1ContractEvent struct {
ContractEvent `gorm:"embedded"`
}
......
......@@ -10,14 +10,15 @@ CREATE TABLE IF NOT EXISTS l1_block_headers (
hash VARCHAR NOT NULL PRIMARY KEY,
parent_hash VARCHAR NOT NULL,
number UINT256,
timestamp INTEGER NOT NULL
timestamp INTEGER NOT NULL CHECK (timestamp > 0)
);
CREATE TABLE IF NOT EXISTS l2_block_headers (
hash VARCHAR NOT NULL PRIMARY KEY,
parent_hash VARCHAR NOT NULL,
number UINT256,
timestamp INTEGER NOT NULL
-- Block header
hash VARCHAR NOT NULL PRIMARY KEY,
parent_hash VARCHAR NOT NULL,
number UINT256,
timestamp INTEGER NOT NULL CHECK (timestamp > 0)
);
/**
......@@ -30,7 +31,7 @@ CREATE TABLE IF NOT EXISTS l1_contract_events (
transaction_hash VARCHAR NOT NULL,
event_signature VARCHAR NOT NULL,
log_index INTEGER NOT NULL,
timestamp INTEGER NOT NULL
timestamp INTEGER NOT NULL CHECK (timestamp > 0)
);
CREATE TABLE IF NOT EXISTS l2_contract_events (
......@@ -39,7 +40,7 @@ CREATE TABLE IF NOT EXISTS l2_contract_events (
transaction_hash VARCHAR NOT NULL,
event_signature VARCHAR NOT NULL,
log_index INTEGER NOT NULL,
timestamp INTEGER NOT NULL
timestamp INTEGER NOT NULL CHECK (timestamp > 0)
);
-- Tables that index finalization markers for L2 blocks.
......@@ -69,6 +70,10 @@ CREATE TABLE IF NOT EXISTS deposits (
-- Event causing the deposit
initiated_l1_event_guid VARCHAR NOT NULL REFERENCES l1_contract_events(guid),
sent_message_nonce UINT256 UNIQUE,
-- Finalization marker for the deposit
finalized_l2_event_guid VARCHAR REFERENCES l2_contract_events(guid),
-- Deposit information (do we need indexes on from/to?)
from_address VARCHAR NOT NULL,
......@@ -78,7 +83,7 @@ CREATE TABLE IF NOT EXISTS deposits (
l2_token_address VARCHAR NOT NULL,
amount UINT256,
data VARCHAR NOT NULL,
timestamp INTEGER NOT NULL
timestamp INTEGER NOT NULL CHECK (timestamp > 0)
);
CREATE TABLE IF NOT EXISTS withdrawals (
......@@ -86,6 +91,7 @@ CREATE TABLE IF NOT EXISTS withdrawals (
-- Event causing this withdrawal
initiated_l2_event_guid VARCHAR NOT NULL REFERENCES l2_contract_events(guid),
sent_message_nonce UINT256 UNIQUE,
-- Multistep (bedrock) process of a withdrawal
withdrawal_hash VARCHAR NOT NULL,
......@@ -101,5 +107,5 @@ CREATE TABLE IF NOT EXISTS withdrawals (
l2_token_address VARCHAR NOT NULL,
amount UINT256,
data VARCHAR NOT NULL,
timestamp INTEGER NOT NULL
timestamp INTEGER NOT NULL CHECK (timestamp > 0)
);
package processor
import (
"errors"
"fmt"
"github.com/ethereum-optimism/optimism/indexer/database"
"github.com/ethereum/go-ethereum/accounts/abi"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/google/uuid"
)
type ProcessedContractEventLogIndexKey struct {
header common.Hash
index uint
}
type ProcessedContractEvents struct {
events []*database.ContractEvent
eventsBySignature map[common.Hash][]*database.ContractEvent
eventByLogIndex map[ProcessedContractEventLogIndexKey]*database.ContractEvent
eventLog map[uuid.UUID]*types.Log
}
func NewProcessedContractEvents() *ProcessedContractEvents {
return &ProcessedContractEvents{
events: []*database.ContractEvent{},
eventsBySignature: make(map[common.Hash][]*database.ContractEvent),
eventByLogIndex: make(map[ProcessedContractEventLogIndexKey]*database.ContractEvent),
eventLog: make(map[uuid.UUID]*types.Log),
}
}
func (p *ProcessedContractEvents) AddLog(log *types.Log, time uint64) *database.ContractEvent {
contractEvent := database.ContractEventFromGethLog(log, time)
p.events = append(p.events, &contractEvent)
p.eventsBySignature[contractEvent.EventSignature] = append(p.eventsBySignature[contractEvent.EventSignature], &contractEvent)
p.eventByLogIndex[ProcessedContractEventLogIndexKey{log.BlockHash, log.Index}] = &contractEvent
p.eventLog[contractEvent.GUID] = log
return &contractEvent
}
func UnpackLog(out interface{}, log *types.Log, name string, contractAbi *abi.ABI) error {
eventAbi, ok := contractAbi.Events[name]
if !ok {
return fmt.Errorf("event %s not present in supplied ABI", name)
} else if len(log.Topics) == 0 {
return errors.New("anonymous events are not supported")
} else if log.Topics[0] != eventAbi.ID {
return errors.New("event signature mismatch")
}
err := contractAbi.UnpackIntoInterface(out, name, log.Data)
if err != nil {
return err
}
// handle topics if present
if len(log.Topics) > 1 {
var indexedArgs abi.Arguments
for _, arg := range eventAbi.Inputs {
if arg.Indexed {
indexedArgs = append(indexedArgs, arg)
}
}
// The first topic (event signature) is omitted
err := abi.ParseTopics(out, indexedArgs, log.Topics[1:])
if err != nil {
return err
}
}
return nil
}
......@@ -122,21 +122,22 @@ func l1ProcessFn(processLog log.Logger, ethClient node.EthClient, l1Contracts L1
headerMap[header.Hash()] = header
}
/** Watch for Contract Events **/
/** Watch for all Optimism Contract Events **/
logFilter := ethereum.FilterQuery{FromBlock: headers[0].Number, ToBlock: headers[numHeaders-1].Number, Addresses: contractAddrs}
logs, err := rawEthClient.FilterLogs(context.Background(), logFilter)
logs, err := rawEthClient.FilterLogs(context.Background(), logFilter) // []types.Log
if err != nil {
return err
}
// L2 checkpoitns posted on L1
// L2 checkpoints posted on L1
outputProposals := []*database.OutputProposal{}
legacyStateBatches := []*database.LegacyStateBatch{}
numLogs := len(logs)
l1ContractEvents := make([]*database.L1ContractEvent, numLogs)
l1HeadersOfInterest := make(map[common.Hash]bool)
l1ContractEvents := make([]*database.L1ContractEvent, len(logs))
processedContractEvents := NewProcessedContractEvents()
for i, log := range logs {
header, ok := headerMap[log.BlockHash]
if !ok {
......@@ -144,19 +145,9 @@ func l1ProcessFn(processLog log.Logger, ethClient node.EthClient, l1Contracts L1
return errors.New("parsed log with a block hash not in this batch")
}
contractEvent := &database.L1ContractEvent{
ContractEvent: database.ContractEvent{
GUID: uuid.New(),
BlockHash: log.BlockHash,
TransactionHash: log.TxHash,
EventSignature: log.Topics[0],
LogIndex: uint64(log.Index),
Timestamp: header.Time,
},
}
l1ContractEvents[i] = contractEvent
contractEvent := processedContractEvents.AddLog(&logs[i], header.Time)
l1HeadersOfInterest[log.BlockHash] = true
l1ContractEvents[i] = &database.L1ContractEvent{ContractEvent: *contractEvent}
// Track Checkpoint Events for L2
switch contractEvent.EventSignature {
......@@ -194,63 +185,188 @@ func l1ProcessFn(processLog log.Logger, ethClient node.EthClient, l1Contracts L1
// we iterate on the original array to maintain ordering. probably can find a more efficient
// way to iterate over the `l1HeadersOfInterest` map while maintaining ordering
l1Headers := []*database.L1BlockHeader{}
indexedL1Headers := []*database.L1BlockHeader{}
for _, header := range headers {
blockHash := header.Hash()
if _, hasLogs := l1HeadersOfInterest[blockHash]; !hasLogs {
_, hasLogs := l1HeadersOfInterest[header.Hash()]
if !hasLogs {
continue
}
l1Headers = append(l1Headers, &database.L1BlockHeader{
BlockHeader: database.BlockHeader{
Hash: blockHash,
ParentHash: header.ParentHash,
Number: database.U256{Int: header.Number},
Timestamp: header.Time,
},
})
indexedL1Headers = append(indexedL1Headers, &database.L1BlockHeader{BlockHeader: database.BlockHeaderFromGethHeader(header)})
}
/** Update Database **/
numL1Headers := len(l1Headers)
if numL1Headers == 0 {
processLog.Info("no l1 blocks of interest")
return nil
numIndexedL1Headers := len(indexedL1Headers)
if numIndexedL1Headers > 0 {
processLog.Info("saving l1 blocks with optimism logs", "size", numIndexedL1Headers, "batch_size", numHeaders)
err = db.Blocks.StoreL1BlockHeaders(indexedL1Headers)
if err != nil {
return err
}
// Since the headers to index are derived from the existence of logs, we know in this branch `numLogs > 0`
processLog.Info("detected contract logs", "size", len(l1ContractEvents))
err = db.ContractEvents.StoreL1ContractEvents(l1ContractEvents)
if err != nil {
return err
}
// Mark L2 checkpoints that have been recorded on L1 (L2OutputProposal & StateBatchAppended events)
numLegacyStateBatches := len(legacyStateBatches)
if numLegacyStateBatches > 0 {
latestBatch := legacyStateBatches[numLegacyStateBatches-1]
latestL2Height := latestBatch.PrevTotal + latestBatch.Size - 1
processLog.Info("detected legacy state batches", "size", numLegacyStateBatches, "latest_l2_block_number", latestL2Height)
}
numOutputProposals := len(outputProposals)
if numOutputProposals > 0 {
latestL2Height := outputProposals[numOutputProposals-1].L2BlockNumber.Int
processLog.Info("detected output proposals", "size", numOutputProposals, "latest_l2_block_number", latestL2Height)
err := db.Blocks.StoreOutputProposals(outputProposals)
if err != nil {
return err
}
}
// forward along contract events to the bridge processor
err = l1BridgeProcessContractEvents(processLog, db, ethClient, processedContractEvents, l1Contracts)
if err != nil {
return err
}
} else {
processLog.Info("no l1 blocks of interest within batch")
}
// a-ok!
return nil
}
}
func l1BridgeProcessContractEvents(processLog log.Logger, db *database.DB, ethClient node.EthClient, events *ProcessedContractEvents, l1Contracts L1Contracts) error {
rawEthClient := ethclient.NewClient(ethClient.RawRpcClient())
// Process New Deposits
initiatedDepositEvents, err := StandardBridgeInitiatedEvents(events)
if err != nil {
return err
}
deposits := make([]*database.Deposit, len(initiatedDepositEvents))
for i, initiatedBridgeEvent := range initiatedDepositEvents {
deposits[i] = &database.Deposit{
GUID: uuid.New(),
InitiatedL1EventGUID: initiatedBridgeEvent.RawEvent.GUID,
SentMessageNonce: database.U256{Int: initiatedBridgeEvent.CrossDomainMessengerNonce},
TokenPair: database.TokenPair{L1TokenAddress: initiatedBridgeEvent.LocalToken, L2TokenAddress: initiatedBridgeEvent.RemoteToken},
Tx: database.Transaction{
FromAddress: initiatedBridgeEvent.From,
ToAddress: initiatedBridgeEvent.To,
Amount: database.U256{Int: initiatedBridgeEvent.Amount},
Data: initiatedBridgeEvent.ExtraData,
Timestamp: initiatedBridgeEvent.RawEvent.Timestamp,
},
}
}
processLog.Info("saving l1 blocks of interest", "size", numL1Headers, "batch_size", numHeaders)
err = db.Blocks.StoreL1BlockHeaders(l1Headers)
if len(deposits) > 0 {
processLog.Info("detected L1StandardBridge deposits", "num", len(deposits))
err := db.Bridge.StoreDeposits(deposits)
if err != nil {
return err
}
}
// Since the headers to index are derived from the existence of logs, we know in this branch `numLogs > 0`
processLog.Info("saving contract logs", "size", numLogs)
err = db.ContractEvents.StoreL1ContractEvents(l1ContractEvents)
// Prove L2 Withdrawals
provenWithdrawalEvents, err := OptimismPortalWithdrawalProvenEvents(events)
if err != nil {
return err
}
// we manually keep track since not every proven withdrawal is a standard bridge withdrawal
numProvenWithdrawals := 0
for _, provenWithdrawalEvent := range provenWithdrawalEvents {
withdrawalHash := provenWithdrawalEvent.WithdrawalHash
withdrawal, err := db.Bridge.WithdrawalByHash(withdrawalHash)
if err != nil {
return err
}
// Mark L2 checkpoints that have been recorded on L1 (L2OutputProposal & StateBatchAppended events)
numLegacyStateBatches := len(legacyStateBatches)
if numLegacyStateBatches > 0 {
latestBatch := legacyStateBatches[numLegacyStateBatches-1]
latestL2Height := latestBatch.PrevTotal + latestBatch.Size - 1
processLog.Info("detected legacy state batches", "size", numLegacyStateBatches, "latest_l2_block_number", latestL2Height)
}
// Check if the L2Processor is behind or really has missed an event. We can compare against the
// OptimismPortal#ProvenWithdrawal on-chain mapping relative to the latest indexed L2 height
if withdrawal == nil {
bridgeAddress := l1Contracts.L1StandardBridge
portalAddress := l1Contracts.OptimismPortal
if provenWithdrawalEvent.From != bridgeAddress || provenWithdrawalEvent.To != bridgeAddress {
// non-bridge withdrawal
continue
}
numOutputProposals := len(outputProposals)
if numOutputProposals > 0 {
latestL2Height := outputProposals[numOutputProposals-1].L2BlockNumber.Int
processLog.Info("detected output proposals", "size", numOutputProposals, "latest_l2_block_number", latestL2Height)
err := db.Blocks.StoreOutputProposals(outputProposals)
// Query for the the proven withdrawal on-chain
provenWithdrawal, err := OptimismPortalQueryProvenWithdrawal(rawEthClient, portalAddress, withdrawalHash)
if err != nil {
return err
}
latestL2Header, err := db.Blocks.LatestL2BlockHeader()
if err != nil {
return err
}
if latestL2Header == nil || provenWithdrawal.L2OutputIndex.Cmp(latestL2Header.Number.Int) > 0 {
processLog.Warn("behind on indexed L2 withdrawals")
return errors.New("waiting for L2Processor to catch up")
} else {
processLog.Crit("missing indexed withdrawal for this proven event")
return errors.New("missing withdrawal message")
}
}
// a-ok!
return nil
err = db.Bridge.MarkProvenWithdrawalEvent(withdrawal.GUID, provenWithdrawalEvent.RawEvent.GUID)
if err != nil {
return err
}
numProvenWithdrawals++
}
if numProvenWithdrawals > 0 {
processLog.Info("proven L2StandardBridge withdrawals", "size", numProvenWithdrawals)
}
// Finalize Pending Withdrawals
finalizedWithdrawalEvents, err := StandardBridgeFinalizedEvents(rawEthClient, events)
if err != nil {
return err
}
for _, finalizedBridgeEvent := range finalizedWithdrawalEvents {
nonce := finalizedBridgeEvent.CrossDomainMessengerNonce
withdrawal, err := db.Bridge.WithdrawalByMessageNonce(nonce)
if err != nil {
processLog.Error("error querying associated withdrawal messsage using nonce", "cross_domain_messenger_nonce", nonce)
return err
}
// Since we have to prove the event on-chain first, we don't need to check if the processor is
// behind. we're definitely in an error state if we cannot find the withdrawal when parsing this even
if withdrawal == nil {
processLog.Crit("missing indexed withdrawal for this finalization event")
return errors.New("missing withdrawal message")
}
err = db.Bridge.MarkFinalizedWithdrawalEvent(withdrawal.GUID, finalizedBridgeEvent.RawEvent.GUID)
if err != nil {
processLog.Error("error finalizing withdrawal", "err", err)
return err
}
}
if len(finalizedWithdrawalEvents) > 0 {
processLog.Info("finalized L2StandardBridge withdrawals", "num", len(finalizedWithdrawalEvents))
}
// a-ok!
return nil
}
......@@ -7,6 +7,7 @@ import (
"github.com/ethereum-optimism/optimism/indexer/database"
"github.com/ethereum-optimism/optimism/indexer/node"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/google/uuid"
"github.com/ethereum/go-ethereum"
......@@ -124,8 +125,8 @@ func l2ProcessFn(processLog log.Logger, ethClient node.EthClient, l2Contracts L2
return err
}
numLogs := len(logs)
l2ContractEvents := make([]*database.L2ContractEvent, numLogs)
l2ContractEvents := make([]*database.L2ContractEvent, len(logs))
processedContractEvents := NewProcessedContractEvents()
for i, log := range logs {
header, ok := l2HeaderMap[log.BlockHash]
if !ok {
......@@ -133,16 +134,8 @@ func l2ProcessFn(processLog log.Logger, ethClient node.EthClient, l2Contracts L2
return errors.New("parsed log with a block hash not in this batch")
}
l2ContractEvents[i] = &database.L2ContractEvent{
ContractEvent: database.ContractEvent{
GUID: uuid.New(),
BlockHash: log.BlockHash,
TransactionHash: log.TxHash,
EventSignature: log.Topics[0],
LogIndex: uint64(log.Index),
Timestamp: header.Time,
},
}
contractEvent := processedContractEvents.AddLog(&logs[i], header.Time)
l2ContractEvents[i] = &database.L2ContractEvent{ContractEvent: *contractEvent}
}
/** Update Database **/
......@@ -153,15 +146,118 @@ func l2ProcessFn(processLog log.Logger, ethClient node.EthClient, l2Contracts L2
return err
}
numLogs := len(l2ContractEvents)
if numLogs > 0 {
processLog.Info("detected contract logs", "size", numLogs)
err = db.ContractEvents.StoreL2ContractEvents(l2ContractEvents)
if err != nil {
return err
}
// forward along contract events to the bridge processor
err = l2BridgeProcessContractEvents(processLog, db, ethClient, processedContractEvents)
if err != nil {
return err
}
}
// a-ok!
return nil
}
}
func l2BridgeProcessContractEvents(processLog log.Logger, db *database.DB, ethClient node.EthClient, events *ProcessedContractEvents) error {
rawEthClient := ethclient.NewClient(ethClient.RawRpcClient())
l2ToL1MessagePasserABI, err := bindings.L2ToL1MessagePasserMetaData.GetAbi()
if err != nil {
return err
}
messagePassedEventAbi := l2ToL1MessagePasserABI.Events["MessagePassed"]
// Process New Withdrawals
initiatedWithdrawalEvents, err := StandardBridgeInitiatedEvents(events)
if err != nil {
return err
}
withdrawals := make([]*database.Withdrawal, len(initiatedWithdrawalEvents))
for i, initiatedBridgeEvent := range initiatedWithdrawalEvents {
log := events.eventLog[initiatedBridgeEvent.RawEvent.GUID]
// extract the withdrawal hash from the MessagePassed event
var msgPassedData bindings.L2ToL1MessagePasserMessagePassed
msgPassedLog := events.eventLog[events.eventByLogIndex[ProcessedContractEventLogIndexKey{log.BlockHash, log.Index + 1}].GUID]
err := UnpackLog(&msgPassedData, msgPassedLog, messagePassedEventAbi.Name, l2ToL1MessagePasserABI)
if err != nil {
return err
}
withdrawals[i] = &database.Withdrawal{
GUID: uuid.New(),
InitiatedL2EventGUID: initiatedBridgeEvent.RawEvent.GUID,
SentMessageNonce: database.U256{Int: initiatedBridgeEvent.CrossDomainMessengerNonce},
WithdrawalHash: msgPassedData.WithdrawalHash,
TokenPair: database.TokenPair{L1TokenAddress: initiatedBridgeEvent.LocalToken, L2TokenAddress: initiatedBridgeEvent.RemoteToken},
Tx: database.Transaction{
FromAddress: initiatedBridgeEvent.From,
ToAddress: initiatedBridgeEvent.To,
Amount: database.U256{Int: initiatedBridgeEvent.Amount},
Data: initiatedBridgeEvent.ExtraData,
Timestamp: initiatedBridgeEvent.RawEvent.Timestamp,
},
}
}
if len(withdrawals) > 0 {
processLog.Info("detected L2StandardBridge withdrawals", "num", len(withdrawals))
err := db.Bridge.StoreWithdrawals(withdrawals)
if err != nil {
return err
}
}
// Finalize Deposits
finalizationBridgeEvents, err := StandardBridgeFinalizedEvents(rawEthClient, events)
if err != nil {
return err
}
for _, finalizedBridgeEvent := range finalizationBridgeEvents {
nonce := finalizedBridgeEvent.CrossDomainMessengerNonce
deposit, err := db.Bridge.DepositByMessageNonce(nonce)
if err != nil {
processLog.Error("error querying associated deposit messsage using nonce", "cross_domain_messenger_nonce", nonce)
return err
} else if deposit == nil {
latestNonce, err := db.Bridge.LatestDepositMessageNonce()
if err != nil {
return err
}
// Check if the L1Processor is behind or really has missed an event
if latestNonce == nil || nonce.Cmp(latestNonce) > 0 {
processLog.Warn("behind on indexed L1 deposits")
return errors.New("waiting for L1Processor to catch up")
} else {
processLog.Crit("missing indexed deposit for this finalization event")
return errors.New("missing deposit message")
}
}
err = db.Bridge.MarkFinalizedDepositEvent(deposit.GUID, finalizedBridgeEvent.RawEvent.GUID)
if err != nil {
processLog.Error("error finalizing deposit", "err", err)
return err
}
}
if len(finalizationBridgeEvents) > 0 {
processLog.Info("finalized L1StandardBridge deposits", "size", len(finalizationBridgeEvents))
}
// a-ok
return nil
}
package processor
import (
"context"
"math/big"
"github.com/ethereum-optimism/optimism/indexer/database"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/ethclient"
)
type OptimismPortalWithdrawalProvenEvent struct {
*bindings.OptimismPortalWithdrawalProven
RawEvent *database.ContractEvent
}
type OptimismPortalProvenWithdrawal struct {
OutputRoot [32]byte
Timestamp *big.Int
L2OutputIndex *big.Int
}
func OptimismPortalWithdrawalProvenEvents(events *ProcessedContractEvents) ([]OptimismPortalWithdrawalProvenEvent, error) {
optimismPortalAbi, err := bindings.OptimismPortalMetaData.GetAbi()
if err != nil {
return nil, err
}
processedWithdrawalProvenEvents := events.eventsBySignature[optimismPortalAbi.Events["WithdrawalProven"].ID]
provenEvents := make([]OptimismPortalWithdrawalProvenEvent, len(processedWithdrawalProvenEvents))
for i, provenEvent := range processedWithdrawalProvenEvents {
provenEvents[i] = OptimismPortalWithdrawalProvenEvent{nil, provenEvent}
}
return provenEvents, nil
}
func OptimismPortalQueryProvenWithdrawal(ethClient *ethclient.Client, portalAddress common.Address, withdrawalHash common.Hash) (OptimismPortalProvenWithdrawal, error) {
var provenWithdrawal OptimismPortalProvenWithdrawal
optimismPortalAbi, err := bindings.OptimismPortalMetaData.GetAbi()
if err != nil {
return provenWithdrawal, err
}
name := "provenWithdrawals"
txData, err := optimismPortalAbi.Pack(name, withdrawalHash)
if err != nil {
return provenWithdrawal, err
}
callMsg := ethereum.CallMsg{To: &portalAddress, Data: txData}
data, err := ethClient.CallContract(context.Background(), callMsg, nil)
if err != nil {
return provenWithdrawal, err
}
err = optimismPortalAbi.UnpackIntoInterface(&provenWithdrawal, name, data)
if err != nil {
return provenWithdrawal, err
}
return provenWithdrawal, nil
}
......@@ -55,8 +55,8 @@ func (p processor) Start() {
firstHeader := unprocessedHeaders[0]
lastHeader := unprocessedHeaders[len(unprocessedHeaders)-1]
batchLog := p.processLog.New("batch_start_block_number", firstHeader.Number, "batch_end_block_number", lastHeader.Number)
batchLog.Info("processing batch")
err := p.db.Transaction(func(db *database.DB) error {
batchLog.Info("processing batch")
return p.processFn(db, unprocessedHeaders)
})
......
package processor
import (
"bytes"
"context"
"errors"
"math/big"
"github.com/ethereum-optimism/optimism/indexer/database"
"github.com/ethereum-optimism/optimism/op-bindings/bindings"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/ethclient"
)
var (
ethAddress = common.HexToAddress("0xDeadDeAddeAddEAddeadDEaDDEAdDeaDDeAD0000")
)
type StandardBridgeInitiatedEvent struct {
// We hardcode to ERC20 since ETH can be pseudo-represented as an ERC20 utilizing
// the hardcoded ETH address
*bindings.L1StandardBridgeERC20BridgeInitiated
CrossDomainMessengerNonce *big.Int
RawEvent *database.ContractEvent
}
type StandardBridgeFinalizedEvent struct {
// We hardcode to ERC20 since ETH can be pseudo-represented as an ERC20 utilizing
// the hardcoded ETH address
*bindings.L1StandardBridgeERC20BridgeFinalized
CrossDomainMessengerNonce *big.Int
RawEvent *database.ContractEvent
}
// StandardBridgeInitiatedEvents extracts all initiated bridge events from the contracts that follow the StandardBridge ABI. The
// correlated CrossDomainMessenger nonce is also parsed from the associated messenger events.
func StandardBridgeInitiatedEvents(events *ProcessedContractEvents) ([]StandardBridgeInitiatedEvent, error) {
ethBridgeInitiatedEvents, err := _standardBridgeInitiatedEvents[bindings.L1StandardBridgeETHBridgeInitiated](events)
if err != nil {
return nil, err
}
erc20BridgeInitiatedEvents, err := _standardBridgeInitiatedEvents[bindings.L1StandardBridgeERC20BridgeInitiated](events)
if err != nil {
return nil, err
}
return append(ethBridgeInitiatedEvents, erc20BridgeInitiatedEvents...), nil
}
// StandardBridgeFinalizedEvents extracts all finalization bridge events from the contracts that follow the StandardBridge ABI. The
// correlated CrossDomainMessenger nonce is also parsed by looking at the parameters of the corresponding relayMessage transaction data.
func StandardBridgeFinalizedEvents(rawEthClient *ethclient.Client, events *ProcessedContractEvents) ([]StandardBridgeFinalizedEvent, error) {
ethBridgeFinalizedEvents, err := _standardBridgeFinalizedEvents[bindings.L1StandardBridgeETHBridgeFinalized](rawEthClient, events)
if err != nil {
return nil, err
}
erc20BridgeFinalizedEvents, err := _standardBridgeFinalizedEvents[bindings.L1StandardBridgeERC20BridgeFinalized](rawEthClient, events)
if err != nil {
return nil, err
}
return append(ethBridgeFinalizedEvents, erc20BridgeFinalizedEvents...), nil
}
// parse out eth or erc20 bridge initiated events
func _standardBridgeInitiatedEvents[BridgeEvent bindings.L1StandardBridgeETHBridgeInitiated | bindings.L1StandardBridgeERC20BridgeInitiated](
events *ProcessedContractEvents,
) ([]StandardBridgeInitiatedEvent, error) {
l1StandardBridgeABI, err := bindings.L1StandardBridgeMetaData.GetAbi()
if err != nil {
return nil, err
}
l1CrossDomainMessengerABI, err := bindings.L1CrossDomainMessengerMetaData.GetAbi()
if err != nil {
return nil, err
}
sentMessageEventAbi := l1CrossDomainMessengerABI.Events["SentMessage"]
var tmp BridgeEvent
var eventName string
var finalizeMethodName string
switch any(tmp).(type) {
case bindings.L1StandardBridgeETHBridgeInitiated:
eventName = "ETHBridgeInitiated"
finalizeMethodName = "finalizeBridgeETH"
case bindings.L1StandardBridgeERC20BridgeInitiated:
eventName = "ERC20BridgeInitiated"
finalizeMethodName = "finalizeBridgeERC20"
default:
panic("should not be here")
}
processedInitiatedBridgeEvents := events.eventsBySignature[l1StandardBridgeABI.Events[eventName].ID]
initiatedBridgeEvents := make([]StandardBridgeInitiatedEvent, len(processedInitiatedBridgeEvents))
for i, bridgeInitiatedEvent := range processedInitiatedBridgeEvents {
log := events.eventLog[bridgeInitiatedEvent.GUID]
bridgeData := new(BridgeEvent)
err := UnpackLog(bridgeData, log, eventName, l1StandardBridgeABI)
if err != nil {
return nil, err
}
// Look for the sent message event to extract the associated messager nonce
// - L1: BridgeInitiated -> Portal#DepositTransaction -> SentMessage ...
// - L1: BridgeInitiated -> L2ToL1MessagePasser#MessagePassed -> SentMessage ...
var sentMsgData bindings.L1CrossDomainMessengerSentMessage
sentMsgLog := events.eventLog[events.eventByLogIndex[ProcessedContractEventLogIndexKey{log.BlockHash, log.Index + 2}].GUID]
err = UnpackLog(&sentMsgData, sentMsgLog, sentMessageEventAbi.Name, l1CrossDomainMessengerABI)
if err != nil {
return nil, err
}
var erc20BridgeData *bindings.L1StandardBridgeERC20BridgeInitiated
var expectedCrossDomainMessage []byte
switch any(bridgeData).(type) {
case *bindings.L1StandardBridgeETHBridgeInitiated:
ethBridgeData := any(bridgeData).(*bindings.L1StandardBridgeETHBridgeInitiated)
expectedCrossDomainMessage, err = l1StandardBridgeABI.Pack(finalizeMethodName, ethBridgeData.From, ethBridgeData.To, ethBridgeData.Amount, ethBridgeData.ExtraData)
if err != nil {
return nil, err
}
// represent eth bridge as an erc20
erc20BridgeData = &bindings.L1StandardBridgeERC20BridgeInitiated{
// Represent ETH using the hardcoded address
LocalToken: ethAddress, RemoteToken: ethAddress,
// Bridge data
From: ethBridgeData.From, To: ethBridgeData.To, Amount: ethBridgeData.Amount, ExtraData: ethBridgeData.ExtraData,
}
case *bindings.L1StandardBridgeERC20BridgeInitiated:
_temp := any(bridgeData).(bindings.L1StandardBridgeERC20BridgeInitiated)
erc20BridgeData = &_temp
expectedCrossDomainMessage, err = l1StandardBridgeABI.Pack(finalizeMethodName, erc20BridgeData.RemoteToken, erc20BridgeData.LocalToken, erc20BridgeData.From, erc20BridgeData.To, erc20BridgeData.Amount, erc20BridgeData.ExtraData)
if err != nil {
return nil, err
}
}
if !bytes.Equal(sentMsgData.Message, expectedCrossDomainMessage) {
return nil, errors.New("bridge cross domain message mismatch")
}
initiatedBridgeEvents[i] = StandardBridgeInitiatedEvent{erc20BridgeData, sentMsgData.MessageNonce, bridgeInitiatedEvent}
}
return initiatedBridgeEvents, nil
}
// parse out eth or erc20 bridge finalization events
func _standardBridgeFinalizedEvents[BridgeEvent bindings.L1StandardBridgeETHBridgeFinalized | bindings.L1StandardBridgeERC20BridgeFinalized](
rawEthClient *ethclient.Client,
events *ProcessedContractEvents,
) ([]StandardBridgeFinalizedEvent, error) {
l1StandardBridgeABI, err := bindings.L1StandardBridgeMetaData.GetAbi()
if err != nil {
return nil, err
}
l1CrossDomainMessengerABI, err := bindings.L1CrossDomainMessengerMetaData.GetAbi()
if err != nil {
return nil, err
}
relayedMessageEventAbi := l1CrossDomainMessengerABI.Events["RelayedMessage"]
relayMessageMethodAbi := l1CrossDomainMessengerABI.Methods["relayMessage"]
var bridgeData BridgeEvent
var eventName string
switch any(bridgeData).(type) {
case bindings.L1StandardBridgeETHBridgeFinalized:
eventName = "ETHBridgeFinalized"
case bindings.L1StandardBridgeERC20BridgeFinalized:
eventName = "ERC20BridgeFinalized"
default:
panic("should not be here")
}
processedFinalizedBridgeEvents := events.eventsBySignature[l1StandardBridgeABI.Events[eventName].ID]
finalizedBridgeEvents := make([]StandardBridgeFinalizedEvent, len(processedFinalizedBridgeEvents))
for i, bridgeFinalizedEvent := range processedFinalizedBridgeEvents {
log := events.eventLog[bridgeFinalizedEvent.GUID]
var bridgeData BridgeEvent
err := UnpackLog(&bridgeData, log, eventName, l1StandardBridgeABI)
if err != nil {
return nil, err
}
// Look for the RelayedMessage event that follows right after the BridgeFinalized Event
relayedMsgLog := events.eventLog[events.eventByLogIndex[ProcessedContractEventLogIndexKey{log.BlockHash, log.Index + 1}].GUID]
if relayedMsgLog.Topics[0] != relayedMessageEventAbi.ID {
return nil, errors.New("unexpected bridge event ordering")
}
// There's no way to extract the nonce on the relayed message event. we can extract
// the nonce by unpacking the transaction input for the `relayMessage` transaction
tx, isPending, err := rawEthClient.TransactionByHash(context.Background(), relayedMsgLog.TxHash)
if err != nil || isPending {
return nil, errors.New("unable to query relayMessage tx for bridge finalization event")
}
txData := tx.Data()
if !bytes.Equal(txData[:4], relayMessageMethodAbi.ID) {
return nil, errors.New("bridge finalization event does not match relayMessage tx invocation")
}
inputsMap := make(map[string]interface{})
err = relayMessageMethodAbi.Inputs.UnpackIntoMap(inputsMap, txData[4:])
if err != nil {
return nil, err
}
nonce, ok := inputsMap["_nonce"].(*big.Int)
if !ok {
return nil, errors.New("unable to extract `_nonce` parameter from relayMessage transaction")
}
var erc20BridgeData *bindings.L1StandardBridgeERC20BridgeFinalized
switch any(bridgeData).(type) {
case bindings.L1StandardBridgeETHBridgeInitiated:
ethBridgeData := any(bridgeData).(bindings.L1StandardBridgeETHBridgeFinalized)
erc20BridgeData = &bindings.L1StandardBridgeERC20BridgeFinalized{
// Represent ETH using the hardcoded address
LocalToken: ethAddress, RemoteToken: ethAddress,
// Bridge data
From: ethBridgeData.From, To: ethBridgeData.To, Amount: ethBridgeData.Amount, ExtraData: ethBridgeData.ExtraData,
}
case bindings.L1StandardBridgeERC20BridgeInitiated:
_temp := any(bridgeData).(bindings.L1StandardBridgeERC20BridgeFinalized)
erc20BridgeData = &_temp
}
finalizedBridgeEvents[i] = StandardBridgeFinalizedEvent{erc20BridgeData, nonce, bridgeFinalizedEvent}
}
return finalizedBridgeEvents, nil
}
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