Add unique constraints to prevent duplicate rows across dedup tables

pkg/server/persistence/beacon_bad_blob.go

@@ -12,20 +12,20 @@ import (
 type BeaconBadBlob struct {
 	gorm.Model
 	ID   string `gorm:"primaryKey"`
-	Node string `gorm:"index;index:idx_beacon_bad_blob_node_slot_blockroot_network_fetchedat_index,where:deleted_at IS NULL,priority:1"`
+	Node string `gorm:"index;index:idx_beacon_bad_blob_node_slot_blockroot_network_fetchedat_index,where:deleted_at IS NULL,priority:1;uniqueIndex:idx_beacon_bad_blob_unique,where:deleted_at IS NULL,priority:1"`
 	// We have to use int64 here as SQLite doesn't support uint64. This sucks
 	// but slot 9223372036854775808 is probably around the heat death
 	// of the universe so we should be OK.
-	Slot                 int64 `gorm:"index:idx_beacon_bad_blob_slot,where:deleted_at IS NULL;index;index:idx_beacon_bad_blob_node_slot_blockroot_network_fetchedat_index,where:deleted_at IS NULL,priority:2"`
+	Slot                 int64 `gorm:"index:idx_beacon_bad_blob_slot,where:deleted_at IS NULL;index;index:idx_beacon_bad_blob_node_slot_blockroot_network_fetchedat_index,where:deleted_at IS NULL,priority:2;uniqueIndex:idx_beacon_bad_blob_unique,where:deleted_at IS NULL,priority:2"`
 	Epoch                int64
-	BlockRoot            string    `gorm:"index;index:idx_beacon_bad_blob_node_slot_blockroot_network_fetchedat_index,where:deleted_at IS NULL,priority:3"`
+	BlockRoot            string    `gorm:"index;index:idx_beacon_bad_blob_node_slot_blockroot_network_fetchedat_index,where:deleted_at IS NULL,priority:3;uniqueIndex:idx_beacon_bad_blob_unique,where:deleted_at IS NULL,priority:3"`
 	FetchedAt            time.Time `gorm:"index;index:idx_beacon_bad_blob_node_slot_blockroot_network_fetchedat_index,where:deleted_at IS NULL,priority:5;index:idx_beacon_bad_blob_fetchedat,where:deleted_at IS NULL;index:idx_beacon_bad_blob_fetchedat_network,where:deleted_at IS NULL,priority:1"`
 	BeaconImplementation string
 	NodeVersion          string `gorm:"not null;default:''"`
 	Location             string `gorm:"not null;default:''"`
 	ContentEncoding      string `gorm:"not null;default:''"`
-	Network              string `gorm:"not null;default:'';index;index:idx_beacon_bad_blob_node_slot_blockroot_network_fetchedat_index,where:deleted_at IS NULL,priority:4;index:idx_beacon_bad_blob_network,where:deleted_at IS NULL;index:idx_beacon_bad_blob_fetchedat_network,where:deleted_at IS NULL,priority:2"`
-	Index                int64  `gorm:"index;index:idx_beacon_bad_blob_node_slot_blockroot_network_fetchedat_index,where:deleted_at IS NULL,priority:6"`
+	Network              string `gorm:"not null;default:'';index;index:idx_beacon_bad_blob_node_slot_blockroot_network_fetchedat_index,where:deleted_at IS NULL,priority:4;index:idx_beacon_bad_blob_network,where:deleted_at IS NULL;index:idx_beacon_bad_blob_fetchedat_network,where:deleted_at IS NULL,priority:2;uniqueIndex:idx_beacon_bad_blob_unique,where:deleted_at IS NULL,priority:4"`
+	Index                int64  `gorm:"index;index:idx_beacon_bad_blob_node_slot_blockroot_network_fetchedat_index,where:deleted_at IS NULL,priority:6;uniqueIndex:idx_beacon_bad_blob_unique,where:deleted_at IS NULL,priority:5"`
 }
 
 type BeaconBadBlobFilter struct {

pkg/server/persistence/beacon_bad_block.go

@@ -12,19 +12,19 @@ import (
 type BeaconBadBlock struct {
 	gorm.Model
 	ID   string `gorm:"primaryKey"`
-	Node string `gorm:"index;index:idx_beacon_bad_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:1"`
+	Node string `gorm:"index;index:idx_beacon_bad_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:1;uniqueIndex:idx_beacon_bad_block_unique,where:deleted_at IS NULL,priority:1"`
 	// We have to use int64 here as SQLite doesn't support uint64. This sucks
 	// but slot 9223372036854775808 is probably around the heat death
 	// of the universe so we should be OK.
-	Slot                 int64 `gorm:"index:idx_beacon_bad_block_slot,where:deleted_at IS NULL;index;index:idx_beacon_bad_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:2"`
+	Slot                 int64 `gorm:"index:idx_beacon_bad_block_slot,where:deleted_at IS NULL;index;index:idx_beacon_bad_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:2;uniqueIndex:idx_beacon_bad_block_unique,where:deleted_at IS NULL,priority:2"`
 	Epoch                int64
-	BlockRoot            string    `gorm:"index;index:idx_beacon_bad_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:3"`
+	BlockRoot            string    `gorm:"index;index:idx_beacon_bad_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:3;uniqueIndex:idx_beacon_bad_block_unique,where:deleted_at IS NULL,priority:3"`
 	FetchedAt            time.Time `gorm:"index;index:idx_beacon_bad_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:5;index:idx_beacon_bad_block_fetchedat,where:deleted_at IS NULL;index:idx_beacon_bad_block_fetchedat_network,where:deleted_at IS NULL,priority:1"`
 	BeaconImplementation string
 	NodeVersion          string `gorm:"not null;default:''"`
 	Location             string `gorm:"not null;default:''"`
 	ContentEncoding      string `gorm:"not null;default:''"`
-	Network              string `gorm:"not null;default:'';index;index:idx_beacon_bad_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:4;index:idx_beacon_bad_block_network,where:deleted_at IS NULL;index:idx_beacon_bad_block_network,where:deleted_at IS NULL;index:idx_beacon_bad_block_fetchedat_network,where:deleted_at IS NULL,priority:2"`
+	Network              string `gorm:"not null;default:'';index;index:idx_beacon_bad_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:4;index:idx_beacon_bad_block_network,where:deleted_at IS NULL;index:idx_beacon_bad_block_network,where:deleted_at IS NULL;index:idx_beacon_bad_block_fetchedat_network,where:deleted_at IS NULL,priority:2;uniqueIndex:idx_beacon_bad_block_unique,where:deleted_at IS NULL,priority:4"`
 }
 
 type BeaconBadBlockFilter struct {

pkg/server/persistence/beacon_block.go

@@ -12,19 +12,19 @@ import (
 type BeaconBlock struct {
 	gorm.Model
 	ID   string `gorm:"primaryKey"`
-	Node string `gorm:"index;index:idx_beacon_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:1"`
+	Node string `gorm:"index;index:idx_beacon_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:1;uniqueIndex:idx_beacon_block_unique,where:deleted_at IS NULL,priority:1"`
 	// We have to use int64 here as SQLite doesn't support uint64. This sucks
 	// but slot 9223372036854775808 is probably around the heat death
 	// of the universe so we should be OK.
-	Slot                 int64 `gorm:"index:idx_beacon_block_slot,where:deleted_at IS NULL;index;index:idx_beacon_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:2"`
+	Slot                 int64 `gorm:"index:idx_beacon_block_slot,where:deleted_at IS NULL;index;index:idx_beacon_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:2;uniqueIndex:idx_beacon_block_unique,where:deleted_at IS NULL,priority:2"`
 	Epoch                int64
-	BlockRoot            string    `gorm:"index;index:idx_beacon_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:3"`
+	BlockRoot            string    `gorm:"index;index:idx_beacon_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:3;uniqueIndex:idx_beacon_block_unique,where:deleted_at IS NULL,priority:3"`
 	FetchedAt            time.Time `gorm:"index;index:idx_beacon_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:5;index:idx_beacon_block_fetchedat,where:deleted_at IS NULL;index:idx_beacon_block_fetchedat_network,where:deleted_at IS NULL,priority:1"`
 	BeaconImplementation string
 	NodeVersion          string `gorm:"not null;default:''"`
 	ContentEncoding      string `gorm:"not null;default:''"`
 	Location             string `gorm:"not null;default:''"`
-	Network              string `gorm:"not null;default:'';index;index:idx_beacon_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:4;index:idx_beacon_block_network,where:deleted_at IS NULL;index:idx_beacon_block_network,where:deleted_at IS NULL;index:idx_beacon_block_fetchedat_network,where:deleted_at IS NULL,priority:2"`
+	Network              string `gorm:"not null;default:'';index;index:idx_beacon_block_node_slot_blockroot_network_fetchedat,where:deleted_at IS NULL,priority:4;index:idx_beacon_block_network,where:deleted_at IS NULL;index:idx_beacon_block_network,where:deleted_at IS NULL;index:idx_beacon_block_fetchedat_network,where:deleted_at IS NULL,priority:2;uniqueIndex:idx_beacon_block_unique,where:deleted_at IS NULL,priority:4"`
 }
 
 type BeaconBlockFilter struct {

pkg/server/persistence/beacon_state.go

@@ -12,19 +12,19 @@ import (
 type BeaconState struct {
 	gorm.Model
 	ID   string `gorm:"primaryKey"`
-	Node string `gorm:"index;index:idx_beacon_state_node_slot_stateroot_network_fetchedat,where:deleted_at IS NULL,priority:1"`
+	Node string `gorm:"index;index:idx_beacon_state_node_slot_stateroot_network_fetchedat,where:deleted_at IS NULL,priority:1;uniqueIndex:idx_beacon_state_unique,where:deleted_at IS NULL,priority:1"`
 	// We have to use int64 here as SQLite doesn't support uint64. This sucks
 	// but slot 9223372036854775808 is probably around the heat death
 	// of the universe so we should be OK.
-	Slot                 int64 `gorm:"index:idx_beacon_state_slot,where:deleted_at IS NULL;index;index:idx_beacon_state_node_slot_stateroot_network_fetchedat,where:deleted_at IS NULL,priority:2"`
+	Slot                 int64 `gorm:"index:idx_beacon_state_slot,where:deleted_at IS NULL;index;index:idx_beacon_state_node_slot_stateroot_network_fetchedat,where:deleted_at IS NULL,priority:2;uniqueIndex:idx_beacon_state_unique,where:deleted_at IS NULL,priority:2"`
 	Epoch                int64
-	StateRoot            string    `gorm:"index;index:idx_beacon_state_node_slot_stateroot_network_fetchedat,where:deleted_at IS NULL,priority:3"`
+	StateRoot            string    `gorm:"index;index:idx_beacon_state_node_slot_stateroot_network_fetchedat,where:deleted_at IS NULL,priority:3;uniqueIndex:idx_beacon_state_unique,where:deleted_at IS NULL,priority:3"`
 	FetchedAt            time.Time `gorm:"index;index:idx_beacon_state_node_slot_stateroot_network_fetchedat,where:deleted_at IS NULL,priority:5;index:idx_beacon_state_fetchedat,where:deleted_at IS NULL;index:idx_beacon_state_fetchedat_network,where:deleted_at IS NULL,priority:1"`
 	BeaconImplementation string
 	NodeVersion          string `gorm:"not null;default:''"`
 	ContentEncoding      string `gorm:"not null;default:''"`
 	Location             string `gorm:"not null;default:''"`
-	Network              string `gorm:"not null;default:'';index;index:idx_beacon_state_node_slot_stateroot_network_fetchedat,where:deleted_at IS NULL,priority:4;index:idx_beacon_state_network,where:deleted_at IS NULL;index:idx_beacon_state_network,where:deleted_at IS NULL;index:idx_beacon_state_fetchedat_network,where:deleted_at IS NULL,priority:2"`
+	Network              string `gorm:"not null;default:'';index;index:idx_beacon_state_node_slot_stateroot_network_fetchedat,where:deleted_at IS NULL,priority:4;index:idx_beacon_state_network,where:deleted_at IS NULL;index:idx_beacon_state_network,where:deleted_at IS NULL;index:idx_beacon_state_fetchedat_network,where:deleted_at IS NULL,priority:2;uniqueIndex:idx_beacon_state_unique,where:deleted_at IS NULL,priority:4"`
 }
 
 type BeaconStateFilter struct {

pkg/server/persistence/db.go

@@ -3,6 +3,8 @@ package persistence
 import (
 	"context"
 	"errors"
+	"fmt"
+	"strings"
 
 	"github.com/glebarez/sqlite"
 	perrors "github.com/pkg/errors"
@@ -70,31 +72,63 @@ func NewIndexer(namespace string, log logrus.FieldLogger, config Config, opts *O
 func (i *Indexer) Start(ctx context.Context) error {
 	i.log.Info("Starting indexer")
 
+	// Each of these tables is gaining a unique constraint on its natural key
+	// as part of this migration. A deployment that has been running for a
+	// while may already have duplicate rows for the same natural key (that
+	// is the exact defect the constraint is being added to prevent), and
+	// AutoMigrate would fail outright trying to create a unique index over
+	// data that violates it. Deduplicating first, keeping the earliest row
+	// per natural key, makes the migration safe to run on an existing
+	// database instead of requiring a manual cleanup before upgrading.
+	if err := i.dedupeBeforeUniqueIndex(&BeaconState{}, "beacon_states", "node", "network", "slot", "state_root"); err != nil {
+		return perrors.Wrap(err, "failed to remove duplicate beacon states")
+	}
+
 	err := i.db.AutoMigrate(&BeaconState{})
 	if err != nil {
 		return perrors.Wrap(err, "failed to auto migrate beacon state")
 	}
 
+	if err := i.dedupeBeforeUniqueIndex(&BeaconBlock{}, "beacon_blocks", "node", "network", "slot", "block_root"); err != nil {
+		return perrors.Wrap(err, "failed to remove duplicate beacon blocks")
+	}
+
 	err = i.db.AutoMigrate(&BeaconBlock{})
 	if err != nil {
 		return perrors.Wrap(err, "failed to auto migrate beacon block")
 	}
 
+	if err := i.dedupeBeforeUniqueIndex(&BeaconBadBlock{}, "beacon_bad_blocks", "node", "network", "slot", "block_root"); err != nil {
+		return perrors.Wrap(err, "failed to remove duplicate beacon bad blocks")
+	}
+
 	err = i.db.AutoMigrate(&BeaconBadBlock{})
 	if err != nil {
 		return perrors.Wrap(err, "failed to auto migrate beacon bad block")
 	}
 
+	if err := i.dedupeBeforeUniqueIndex(&BeaconBadBlob{}, "beacon_bad_blobs", "node", "network", "slot", "block_root", "index"); err != nil {
+		return perrors.Wrap(err, "failed to remove duplicate beacon bad blobs")
+	}
+
 	err = i.db.AutoMigrate(&BeaconBadBlob{})
 	if err != nil {
 		return perrors.Wrap(err, "failed to auto migrate beacon bad blob")
 	}
 
+	if err := i.dedupeBeforeUniqueIndex(&ExecutionBlockTrace{}, "execution_block_traces", "node", "network", "block_hash"); err != nil {
+		return perrors.Wrap(err, "failed to remove duplicate execution block traces")
+	}
+
 	err = i.db.AutoMigrate(&ExecutionBlockTrace{})
 	if err != nil {
 		return perrors.Wrap(err, "failed to auto migrate execution block trace")
 	}
 
+	if err := i.dedupeBeforeUniqueIndex(&ExecutionBadBlock{}, "execution_bad_blocks", "node", "network", "block_hash"); err != nil {
+		return perrors.Wrap(err, "failed to remove duplicate execution bad blocks")
+	}
+
 	err = i.db.AutoMigrate(&ExecutionBadBlock{})
 	if err != nil {
 		return perrors.Wrap(err, "failed to auto migrate execution bad block")
@@ -105,6 +139,10 @@ func (i *Indexer) Start(ctx context.Context) error {
 		return perrors.Wrap(err, "failed to auto migrate distributed lock")
 	}
 
+	if err := i.dedupeBeforeUniqueIndex(&PermanentBlock{}, "permanent_blocks", "block_root", "network"); err != nil {
+		return perrors.Wrap(err, "failed to remove duplicate permanent blocks")
+	}
+
 	err = i.db.AutoMigrate(&PermanentBlock{})
 	if err != nil {
 		return perrors.Wrap(err, "failed to auto migrate permanent block")
@@ -113,6 +151,44 @@ func (i *Indexer) Start(ctx context.Context) error {
 	return nil
 }
 
+// dedupeBeforeUniqueIndex removes all but the earliest row (by created_at,
+// falling back to id for a deterministic tiebreak) within each group of rows
+// that share the given natural-key columns. It is a no-op if the table
+// doesn't exist yet, since a fresh database has nothing to deduplicate.
+func (i *Indexer) dedupeBeforeUniqueIndex(dst interface{}, table string, naturalKeyColumns ...string) error {
+	if !i.db.Migrator().HasTable(dst) {
+		return nil
+	}
+
+	quotedColumns := make([]string, len(naturalKeyColumns))
+	for idx, column := range naturalKeyColumns {
+		quotedColumns[idx] = `"` + column + `"`
+	}
+
+	query := fmt.Sprintf(
+		`DELETE FROM "%s" WHERE "deleted_at" IS NULL AND "id" NOT IN (`+
+			`SELECT "id" FROM (`+
+			`SELECT "id", ROW_NUMBER() OVER (PARTITION BY %s ORDER BY "created_at" ASC, "id" ASC) AS row_num `+
+			`FROM "%s" WHERE "deleted_at" IS NULL`+
+			`) ranked WHERE row_num = 1)`,
+		table, strings.Join(quotedColumns, ", "), table,
+	)
+
+	result := i.db.Exec(query)
+	if result.Error != nil {
+		return result.Error
+	}
+
+	if result.RowsAffected > 0 {
+		i.log.WithFields(logrus.Fields{
+			"table": table,
+			"rows":  result.RowsAffected,
+		}).Warn("Removed duplicate rows before adding a unique constraint")
+	}
+
+	return nil
+}
+
 func (i *Indexer) Stop(ctx context.Context) error {
 	i.log.Info("Stopping indexer")
 

pkg/server/persistence/errors.go

@@ -0,0 +1,21 @@
+package persistence
+
+import "strings"
+
+// IsUniqueConstraintError reports whether err was caused by a unique
+// constraint violation, checking for the error signatures produced by both
+// database backends this package supports (SQLite and Postgres). Neither
+// gorm nor either underlying driver exposes a single driver-agnostic type
+// for this, so this checks for the stable, well-known substrings each
+// database actually produces in its error message rather than depending on
+// either driver's internal error types.
+func IsUniqueConstraintError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	msg := err.Error()
+
+	return strings.Contains(msg, "UNIQUE constraint failed") || // SQLite
+		strings.Contains(msg, "duplicate key value violates unique constraint") // Postgres
+}

pkg/server/persistence/execution_bad_block.go

@@ -13,14 +13,14 @@ import (
 type ExecutionBadBlock struct {
 	gorm.Model
 	ID                      string    `gorm:"primaryKey"`
-	Node                    string    `gorm:"index;index:iidx_execution_bad_block_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:1"`
+	Node                    string    `gorm:"index;index:iidx_execution_bad_block_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:1;uniqueIndex:idx_execution_bad_block_unique,where:deleted_at IS NULL,priority:1"`
 	FetchedAt               time.Time `gorm:"index;index:iidx_execution_bad_block_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:3;index:iidx_execution_bad_block_fetchedat,where:deleted_at IS NULL;index:iidx_execution_bad_block_fetchedat_network,where:deleted_at IS NULL,priority:1"`
 	ExecutionImplementation string
 	NodeVersion             string `gorm:"not null;default:''"`
 	ContentEncoding         string `gorm:"not null;default:''"`
 	Location                string `gorm:"not null;default:''"`
-	Network                 string `gorm:"not null;default:'';index;index:iidx_execution_bad_block_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:4;index:iidx_execution_bad_block_network,where:deleted_at IS NULL;index:iidx_execution_bad_block_fetchedat_network,where:deleted_at IS NULL,priority:2"`
-	BlockHash               string `gorm:"not null;default:'';index;index:iidx_execution_bad_block_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:2"`
+	Network                 string `gorm:"not null;default:'';index;index:iidx_execution_bad_block_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:4;index:iidx_execution_bad_block_network,where:deleted_at IS NULL;index:iidx_execution_bad_block_fetchedat_network,where:deleted_at IS NULL,priority:2;uniqueIndex:idx_execution_bad_block_unique,where:deleted_at IS NULL,priority:3"`
+	BlockHash               string `gorm:"not null;default:'';index;index:iidx_execution_bad_block_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:2;uniqueIndex:idx_execution_bad_block_unique,where:deleted_at IS NULL,priority:2"`
 	BlockNumber             sql.NullInt64
 	BlockExtraData          sql.NullString
 }