database idempotency

indexer/cmd/indexer/main.go

@@ -27,5 +27,6 @@ func main() {
 	app := newCli(GitCommit, GitDate)
 	if err := app.RunContext(ctx, os.Args); err != nil {
 		log.Error("application failed", "err", err)
+		os.Exit(1)
 	}
 }

indexer/database/blocks.go

@@ -7,8 +7,10 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
 
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )
 
 /**
@@ -67,17 +69,23 @@ type BlocksDB interface {
  */
 
 type blocksDB struct {
+	log  log.Logger
 	gorm *gorm.DB
 }
 
-func newBlocksDB(db *gorm.DB) BlocksDB {
-	return &blocksDB{gorm: db}
+func newBlocksDB(log log.Logger, db *gorm.DB) BlocksDB {
+	return &blocksDB{log: log.New("table", "blocks"), gorm: db}
 }
 
 // L1
 
 func (db *blocksDB) StoreL1BlockHeaders(headers []L1BlockHeader) error {
-	result := db.gorm.CreateInBatches(&headers, batchInsertSize)
+	deduped := db.gorm.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "hash"}}, DoNothing: true})
+	result := deduped.CreateInBatches(&headers, batchInsertSize)
+	if result.Error != nil && int(result.RowsAffected) < len(headers) {
+		db.log.Warn("ignored L1 block duplicates", "duplicates", len(headers)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 
@@ -115,7 +123,12 @@ func (db *blocksDB) L1LatestBlockHeader() (*L1BlockHeader, error) {
 // L2
 
 func (db *blocksDB) StoreL2BlockHeaders(headers []L2BlockHeader) error {
-	result := db.gorm.CreateInBatches(&headers, batchInsertSize)
+	deduped := db.gorm.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "hash"}}, DoNothing: true})
+	result := deduped.CreateInBatches(&headers, batchInsertSize)
+	if result.Error != nil && int(result.RowsAffected) < len(headers) {
+		db.log.Warn("ignored L2 block duplicates", "duplicates", len(headers)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 

indexer/database/bridge_messages.go

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

indexer/database/bridge_transactions.go

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

indexer/database/bridge_transfers.go

@@ -5,9 +5,11 @@ import (
 	"fmt"
 
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 
 	"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/log"
 )
 
 var (
@@ -79,11 +81,12 @@ type BridgeTransfersDB interface {
  */
 
 type bridgeTransfersDB struct {
+	log  log.Logger
 	gorm *gorm.DB
 }
 
-func newBridgeTransfersDB(db *gorm.DB) BridgeTransfersDB {
-	return &bridgeTransfersDB{gorm: db}
+func newBridgeTransfersDB(log log.Logger, db *gorm.DB) BridgeTransfersDB {
+	return &bridgeTransfersDB{log: log.New("table", "transfers"), gorm: db}
 }
 
 /**
@@ -91,7 +94,12 @@ func newBridgeTransfersDB(db *gorm.DB) BridgeTransfersDB {
  */
 
 func (db *bridgeTransfersDB) StoreL1BridgeDeposits(deposits []L1BridgeDeposit) error {
-	result := db.gorm.CreateInBatches(&deposits, batchInsertSize)
+	deduped := db.gorm.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "transaction_source_hash"}}, DoNothing: true})
+	result := deduped.CreateInBatches(&deposits, batchInsertSize)
+	if result.Error != nil && int(result.RowsAffected) < len(deposits) {
+		db.log.Warn("ignored L1 bridge transfer duplicates", "duplicates", len(deposits)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 
@@ -204,7 +212,12 @@ l1_bridge_deposits.timestamp, cross_domain_message_hash, local_token_address, re
  */
 
 func (db *bridgeTransfersDB) StoreL2BridgeWithdrawals(withdrawals []L2BridgeWithdrawal) error {
-	result := db.gorm.CreateInBatches(&withdrawals, batchInsertSize)
+	deduped := db.gorm.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "transaction_withdrawal_hash"}}, DoNothing: true})
+	result := deduped.CreateInBatches(&withdrawals, batchInsertSize)
+	if result.Error != nil && int(result.RowsAffected) < len(withdrawals) {
+		db.log.Warn("ignored L2 bridge transfer duplicates", "duplicates", len(withdrawals)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 

indexer/database/contract_events.go

@@ -6,9 +6,11 @@ import (
 	"math/big"
 
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
 
 	"github.com/google/uuid"
 )
@@ -99,17 +101,25 @@ type ContractEventsDB interface {
  */
 
 type contractEventsDB struct {
+	log  log.Logger
 	gorm *gorm.DB
 }
 
-func newContractEventsDB(db *gorm.DB) ContractEventsDB {
-	return &contractEventsDB{gorm: db}
+func newContractEventsDB(log log.Logger, db *gorm.DB) ContractEventsDB {
+	return &contractEventsDB{log: log.New("table", "events"), gorm: db}
 }
 
 // L1
 
 func (db *contractEventsDB) StoreL1ContractEvents(events []L1ContractEvent) error {
-	result := db.gorm.CreateInBatches(&events, batchInsertSize)
+	// Since the block hash refers back to L1, we dont necessarily have to check
+	// that the RLP bytes match when doing conflict resolution.
+	deduped := db.gorm.Clauses(clause.OnConflict{OnConstraint: "l1_contract_events_block_hash_log_index_key", DoNothing: true})
+	result := deduped.CreateInBatches(&events, batchInsertSize)
+	if result.Error != nil && int(result.RowsAffected) < len(events) {
+		db.log.Warn("ignored L1 contract event duplicates", "duplicates", len(events)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 
@@ -176,7 +186,14 @@ func (db *contractEventsDB) L1LatestContractEventWithFilter(filter ContractEvent
 // L2
 
 func (db *contractEventsDB) StoreL2ContractEvents(events []L2ContractEvent) error {
-	result := db.gorm.CreateInBatches(&events, batchInsertSize)
+	// Since the block hash refers back to L2, we dont necessarily have to check
+	// that the RLP bytes match when doing conflict resolution.
+	deduped := db.gorm.Clauses(clause.OnConflict{OnConstraint: "l2_contract_events_block_hash_log_index_key", DoNothing: true})
+	result := deduped.CreateInBatches(&events, batchInsertSize)
+	if result.Error != nil && int(result.RowsAffected) < len(events) {
+		db.log.Warn("ignored L2 contract event duplicates", "duplicates", len(events)-int(result.RowsAffected))
+	}
+
 	return result.Error
 }
 

indexer/database/db.go

@@ -10,6 +10,7 @@ import (
 	"github.com/ethereum-optimism/optimism/indexer/config"
 	_ "github.com/ethereum-optimism/optimism/indexer/database/serializers"
 	"github.com/ethereum-optimism/optimism/op-service/retry"
+
 	"github.com/pkg/errors"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -28,6 +29,7 @@ var (
 
 type DB struct {
 	gorm *gorm.DB
+	log  log.Logger
 
 	Blocks             BlocksDB
 	ContractEvents     ContractEventsDB
@@ -37,7 +39,7 @@ type DB struct {
 }
 
 func NewDB(log log.Logger, dbConfig config.DBConfig) (*DB, error) {
-	retryStrategy := &retry.ExponentialStrategy{Min: 1000, Max: 20_000, MaxJitter: 250}
+	log = log.New("module", "db")
 
 	dsn := fmt.Sprintf("host=%s dbname=%s sslmode=disable", dbConfig.Host, dbConfig.Name)
 	if dbConfig.Port != 0 {
@@ -56,6 +58,7 @@ func NewDB(log log.Logger, dbConfig config.DBConfig) (*DB, error) {
 		Logger:                 newLogger(log),
 	}
 
+	retryStrategy := &retry.ExponentialStrategy{Min: 1000, Max: 20_000, MaxJitter: 250}
 	gorm, err := retry.Do[*gorm.DB](context.Background(), 10, retryStrategy, func() (*gorm.DB, error) {
 		gorm, err := gorm.Open(postgres.Open(dsn), &gormConfig)
 		if err != nil {
@@ -66,16 +69,17 @@ func NewDB(log log.Logger, dbConfig config.DBConfig) (*DB, error) {
 	})
 
 	if err != nil {
-		return nil, fmt.Errorf("failed to connect to database after multiple retries: %w", err)
+		return nil, err
 	}
 
 	db := &DB{
 		gorm:               gorm,
-		Blocks:             newBlocksDB(gorm),
-		ContractEvents:     newContractEventsDB(gorm),
-		BridgeTransfers:    newBridgeTransfersDB(gorm),
-		BridgeMessages:     newBridgeMessagesDB(gorm),
-		BridgeTransactions: newBridgeTransactionsDB(gorm),
+		log:                log,
+		Blocks:             newBlocksDB(log, gorm),
+		ContractEvents:     newContractEventsDB(log, gorm),
+		BridgeTransfers:    newBridgeTransfersDB(log, gorm),
+		BridgeMessages:     newBridgeMessagesDB(log, gorm),
+		BridgeTransactions: newBridgeTransactionsDB(log, gorm),
 	}
 
 	return db, nil
@@ -85,7 +89,16 @@ func NewDB(log log.Logger, dbConfig config.DBConfig) (*DB, error) {
 // transaction. If the supplied function errors, the transaction is rolled back.
 func (db *DB) Transaction(fn func(db *DB) error) error {
 	return db.gorm.Transaction(func(tx *gorm.DB) error {
-		return fn(dbFromGormTx(tx))
+		txDB := &DB{
+			gorm:               tx,
+			Blocks:             newBlocksDB(db.log, tx),
+			ContractEvents:     newContractEventsDB(db.log, tx),
+			BridgeTransfers:    newBridgeTransfersDB(db.log, tx),
+			BridgeMessages:     newBridgeMessagesDB(db.log, tx),
+			BridgeTransactions: newBridgeTransactionsDB(db.log, tx),
+		}
+
+		return fn(txDB)
 	})
 }
 
@@ -98,17 +111,6 @@ func (db *DB) Close() error {
 	return sql.Close()
 }
 
-func dbFromGormTx(tx *gorm.DB) *DB {
-	return &DB{
-		gorm:               tx,
-		Blocks:             newBlocksDB(tx),
-		ContractEvents:     newContractEventsDB(tx),
-		BridgeTransfers:    newBridgeTransfersDB(tx),
-		BridgeMessages:     newBridgeMessagesDB(tx),
-		BridgeTransactions: newBridgeTransactionsDB(tx),
-	}
-}
-
 func (db *DB) ExecuteSQLMigration(migrationsFolder string) error {
 	err := filepath.Walk(migrationsFolder, func(path string, info os.FileInfo, err error) error {
 		// Check for any walking error

indexer/database/logger.go

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

indexer/migrations/20230523_create_schema.sql

@@ -63,6 +63,7 @@ CREATE INDEX IF NOT EXISTS l1_contract_events_timestamp ON l1_contract_events(ti
 CREATE INDEX IF NOT EXISTS l1_contract_events_block_hash ON l1_contract_events(block_hash);
 CREATE INDEX IF NOT EXISTS l1_contract_events_event_signature ON l1_contract_events(event_signature);
 CREATE INDEX IF NOT EXISTS l1_contract_events_contract_address ON l1_contract_events(contract_address);
+ALTER TABLE l1_contract_events ADD UNIQUE (block_hash, log_index);
 
 CREATE TABLE IF NOT EXISTS l2_contract_events (
     -- Searchable fields
@@ -81,6 +82,7 @@ CREATE INDEX IF NOT EXISTS l2_contract_events_timestamp ON l2_contract_events(ti
 CREATE INDEX IF NOT EXISTS l2_contract_events_block_hash ON l2_contract_events(block_hash);
 CREATE INDEX IF NOT EXISTS l2_contract_events_event_signature ON l2_contract_events(event_signature);
 CREATE INDEX IF NOT EXISTS l2_contract_events_contract_address ON l2_contract_events(contract_address);
+ALTER TABLE l2_contract_events ADD UNIQUE (block_hash, log_index);
 
 /**
  * BRIDGING DATA