diff --git a/pkg/server/persistence/beacon_bad_blob.go b/pkg/server/persistence/beacon_bad_blob.go index b09a235..c351d37 100644 --- a/pkg/server/persistence/beacon_bad_blob.go +++ b/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 { diff --git a/pkg/server/persistence/beacon_bad_block.go b/pkg/server/persistence/beacon_bad_block.go index a87dc87..586f170 100644 --- a/pkg/server/persistence/beacon_bad_block.go +++ b/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 { diff --git a/pkg/server/persistence/beacon_block.go b/pkg/server/persistence/beacon_block.go index efc2f60..e0b9a56 100644 --- a/pkg/server/persistence/beacon_block.go +++ b/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 { diff --git a/pkg/server/persistence/beacon_state.go b/pkg/server/persistence/beacon_state.go index c536573..5fe6b1a 100644 --- a/pkg/server/persistence/beacon_state.go +++ b/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 { diff --git a/pkg/server/persistence/db.go b/pkg/server/persistence/db.go index 738523d..d9a80b9 100644 --- a/pkg/server/persistence/db.go +++ b/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") diff --git a/pkg/server/persistence/errors.go b/pkg/server/persistence/errors.go new file mode 100644 index 0000000..13ec101 --- /dev/null +++ b/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 +} diff --git a/pkg/server/persistence/execution_bad_block.go b/pkg/server/persistence/execution_bad_block.go index e63a70b..f63d069 100644 --- a/pkg/server/persistence/execution_bad_block.go +++ b/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 } diff --git a/pkg/server/persistence/execution_block_trace.go b/pkg/server/persistence/execution_block_trace.go index f8e6961..b50cd98 100644 --- a/pkg/server/persistence/execution_block_trace.go +++ b/pkg/server/persistence/execution_block_trace.go @@ -12,14 +12,14 @@ import ( type ExecutionBlockTrace struct { gorm.Model ID string `gorm:"primaryKey"` - Node string `gorm:"index;index:idx_execution_block_trace_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:1"` + Node string `gorm:"index;index:idx_execution_block_trace_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:1;uniqueIndex:idx_execution_block_trace_unique,where:deleted_at IS NULL,priority:1"` FetchedAt time.Time `gorm:"index;index:idx_execution_block_trace_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:3;index:idx_execution_block_trace_fetchedat,where:deleted_at IS NULL;index:idx_execution_block_trace_fetchedat_network,where:deleted_at IS NULL,priority:1"` ExecutionImplementation 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_execution_block_trace_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:4;index:idx_execution_block_trace_network,where:deleted_at IS NULL;index:idx_execution_block_trace_fetchedat_network,where:deleted_at IS NULL,priority:2"` - BlockHash string `gorm:"not null;default:'';index;index:idx_execution_block_trace_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:2"` + Network string `gorm:"not null;default:'';index;index:idx_execution_block_trace_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:4;index:idx_execution_block_trace_network,where:deleted_at IS NULL;index:idx_execution_block_trace_fetchedat_network,where:deleted_at IS NULL,priority:2;uniqueIndex:idx_execution_block_trace_unique,where:deleted_at IS NULL,priority:3"` + BlockHash string `gorm:"not null;default:'';index;index:idx_execution_block_trace_node_blockhash_fetchedat_network,where:deleted_at IS NULL,priority:2;uniqueIndex:idx_execution_block_trace_unique,where:deleted_at IS NULL,priority:2"` BlockNumber int64 } diff --git a/pkg/server/persistence/permanent_block.go b/pkg/server/persistence/permanent_block.go index 64a5882..b0ee95b 100644 --- a/pkg/server/persistence/permanent_block.go +++ b/pkg/server/persistence/permanent_block.go @@ -15,8 +15,8 @@ type PermanentBlock struct { gorm.Model // We have to use int64 here as SQLite doesn't support uint64 Slot int64 `gorm:"index:idx_permanent_block_slot,where:deleted_at IS NULL;index:idx_permanent_block_slot_blockroot_network,where:deleted_at IS NULL,priority:1"` - BlockRoot string `gorm:"index:idx_permanent_block_blockroot,where:deleted_at IS NULL;index:idx_permanent_block_slot_blockroot_network,where:deleted_at IS NULL,priority:2"` - Network string `gorm:"index:idx_permanent_block_network,where:deleted_at IS NULL;index:idx_permanent_block_slot_blockroot_network,where:deleted_at IS NULL,priority:3"` + BlockRoot string `gorm:"index:idx_permanent_block_blockroot,where:deleted_at IS NULL;index:idx_permanent_block_slot_blockroot_network,where:deleted_at IS NULL,priority:2;uniqueIndex:idx_permanent_block_unique,where:deleted_at IS NULL,priority:1"` + Network string `gorm:"index:idx_permanent_block_network,where:deleted_at IS NULL;index:idx_permanent_block_slot_blockroot_network,where:deleted_at IS NULL,priority:3;uniqueIndex:idx_permanent_block_unique,where:deleted_at IS NULL,priority:2"` } type PermanentBlockFilter struct { diff --git a/pkg/server/persistence/unique_constraint_migration_test.go b/pkg/server/persistence/unique_constraint_migration_test.go new file mode 100644 index 0000000..20f9594 --- /dev/null +++ b/pkg/server/persistence/unique_constraint_migration_test.go @@ -0,0 +1,176 @@ +package persistence_test + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + "github.com/ethpandaops/tracoor/pkg/server/persistence" + "github.com/glebarez/sqlite" + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/require" + "gorm.io/gorm" +) + +// legacyBeaconState mirrors BeaconState's shape from before this change -- +// same table, same columns, but no unique constraint -- so it can be used to +// stand up a database that looks exactly like one a pre-fix version of this +// code would have produced, via gorm's own migrator rather than a hand +// written CREATE TABLE that would need to be kept in sync by hand. +type legacyBeaconState struct { + gorm.Model + ID string `gorm:"primaryKey"` + Node string + Slot int64 + Epoch int64 + StateRoot string + FetchedAt time.Time + 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:''"` +} + +func (legacyBeaconState) TableName() string { return "beacon_states" } + +func newFileBackedTestDB(t *testing.T) (string, func()) { + t.Helper() + + dbFile, err := os.CreateTemp("", "unique_constraint_migration_*.db") + require.NoError(t, err) + + dbPath := dbFile.Name() + dbFile.Close() + os.Remove(dbPath) + + cleanup := func() { + os.Remove(dbPath) + os.Remove(dbPath + "-wal") + os.Remove(dbPath + "-shm") + } + + return dbPath, cleanup +} + +// TestMigration_SucceedsOnFreshDatabase is the baseline: a database that has +// never been migrated before must still start up cleanly. The dedup step +// must not error out just because the tables it's looking for don't exist +// yet. +func TestMigration_SucceedsOnFreshDatabase(t *testing.T) { + dbPath, cleanup := newFileBackedTestDB(t) + defer cleanup() + + idx, err := persistence.NewIndexer("migration-test", logrus.New(), persistence.Config{ + DSN: fmt.Sprintf("file:%s?parseTime=True", dbPath), + DriverName: "sqlite", + }, persistence.DefaultOptions().SetMetricsEnabled(false)) + require.NoError(t, err) + + require.NoError(t, idx.Start(context.Background())) +} + +// TestMigration_DeduplicatesExistingRowsBeforeAddingConstraint is the +// highest-stakes test for this change: a database that already has +// duplicate rows -- exactly what the pre-existing TOCTOU race and the +// execution handlers' missing dedup check are known to produce -- must not +// fail to start when it upgrades to a version that adds a unique +// constraint. If this broke, every deployment that had ever hit the +// original race would fail to start after upgrading. +func TestMigration_DeduplicatesExistingRowsBeforeAddingConstraint(t *testing.T) { + dbPath, cleanup := newFileBackedTestDB(t) + defer cleanup() + + ctx := context.Background() + + // Seed the table exactly as the PRE-this-change schema would have left + // it: created via gorm's own migrator against the legacy (no unique + // constraint) shape, with duplicate rows already present, standing in + // for what the original TOCTOU race actually produced in a real + // deployment over time. This can't be done by calling idx.Start() + // first, since that already runs the current (patched) migration -- the + // whole point of this test is to simulate a database that predates the + // fix. + legacyDB, err := gorm.Open(sqlite.Open(fmt.Sprintf("file:%s?parseTime=True", dbPath)), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, legacyDB.AutoMigrate(&legacyBeaconState{})) + + now := time.Now() + + for i := 0; i < 3; i++ { + require.NoError(t, legacyDB.Create(&legacyBeaconState{ + ID: fmt.Sprintf("dup-%d", i), + Node: "race-node", + Network: "mainnet", + Slot: 100, + Epoch: 3, + StateRoot: "0xduplicated", + FetchedAt: now.Add(time.Duration(i) * time.Millisecond), + BeaconImplementation: "teku", + NodeVersion: "1.0.0", + Location: "beacon_state/dup.ssz", + }).Error) + } + + require.NoError(t, legacyDB.Create(&legacyBeaconState{ + ID: "distinct-row", + Node: "other-node", + Network: "mainnet", + Slot: 200, + Epoch: 6, + StateRoot: "0xdistinct", + FetchedAt: now, + BeaconImplementation: "teku", + NodeVersion: "1.0.0", + Location: "beacon_state/distinct.ssz", + }).Error) + + rawConn, err := legacyDB.DB() + require.NoError(t, err) + require.NoError(t, rawConn.Close()) + + // Startup with the current (patched) code, against this pre-existing, + // duplicate-containing database. This is the exact code path a real + // deployment hits on upgrade. + idx2, err := persistence.NewIndexer("migration-test-upgrade", logrus.New(), persistence.Config{ + DSN: fmt.Sprintf("file:%s?parseTime=True", dbPath), + DriverName: "sqlite", + }, persistence.DefaultOptions().SetMetricsEnabled(false)) + require.NoError(t, err) + + err = idx2.Start(ctx) + require.NoError(t, err, "migration must succeed even when pre-existing duplicate rows are present") + + // Exactly one of the three duplicates must remain, and it must be the + // earliest one (deterministic, not arbitrary). + remaining, err := idx2.ListBeaconState(ctx, &persistence.BeaconStateFilter{Node: strPtr("race-node")}, nil) + require.NoError(t, err) + require.Len(t, remaining, 1, "expected exactly one surviving row per natural key after dedup") + require.Equal(t, "dup-0", remaining[0].ID, "expected the earliest row (by FetchedAt/insertion order) to survive") + + // The unrelated, non-duplicated row must be completely untouched. + distinct, err := idx2.ListBeaconState(ctx, &persistence.BeaconStateFilter{ID: strPtr("distinct-row")}, nil) + require.NoError(t, err) + require.Len(t, distinct, 1, "the non-duplicated row must survive untouched") + + // And the constraint must now actually be enforced: a fresh attempt to + // insert a fourth duplicate must fail. + dupErr := idx2.InsertBeaconState(ctx, &persistence.BeaconState{ + ID: "dup-attempt-after-migration", + Node: "race-node", + Network: "mainnet", + Slot: 100, + Epoch: 3, + StateRoot: "0xduplicated", + FetchedAt: now, + BeaconImplementation: "teku", + NodeVersion: "1.0.0", + Location: "beacon_state/dup.ssz", + }) + require.Error(t, dupErr, "expected the unique constraint to reject a new duplicate after migration") + require.True(t, persistence.IsUniqueConstraintError(dupErr)) +} + +func strPtr(s string) *string { return &s } diff --git a/pkg/server/service/indexer/indexer.go b/pkg/server/service/indexer/indexer.go index 108e876..8022480 100644 --- a/pkg/server/service/indexer/indexer.go +++ b/pkg/server/service/indexer/indexer.go @@ -229,6 +229,10 @@ func (i *Indexer) CreateBeaconState(ctx context.Context, req *indexer.CreateBeac } if err := i.db.InsertBeaconState(ctx, ProtoBeaconStateToDBBeaconState(state)); err != nil { + if persistence.IsUniqueConstraintError(err) { + return nil, status.Error(codes.AlreadyExists, "beacon state already indexed") + } + i.log.WithError(err).WithFields(logFields).Error("Failed to index state") return nil, status.Error(codes.Internal, "failed to index state") @@ -487,6 +491,10 @@ func (i *Indexer) CreateBeaconBlock(ctx context.Context, req *indexer.CreateBeac } if err := i.db.InsertBeaconBlock(ctx, ProtoBeaconBlockToDBBeaconBlock(block)); err != nil { + if persistence.IsUniqueConstraintError(err) { + return nil, status.Error(codes.AlreadyExists, "beacon block already indexed") + } + i.log.WithError(err).WithFields(logFields).Error("Failed to index block") return nil, status.Error(codes.Internal, "failed to index block") @@ -753,6 +761,10 @@ func (i *Indexer) CreateBeaconBadBlock(ctx context.Context, req *indexer.CreateB } if err := i.db.InsertBeaconBadBlock(ctx, ProtoBeaconBadBlockToDBBeaconBadBlock(badBlock)); err != nil { + if persistence.IsUniqueConstraintError(err) { + return nil, status.Error(codes.AlreadyExists, "beacon block already indexed") + } + i.log.WithError(err).WithFields(logFields).Error("Failed to index bad block") return nil, status.Error(codes.Internal, "failed to index bad block") @@ -1014,6 +1026,10 @@ func (i *Indexer) CreateBeaconBadBlob(ctx context.Context, req *indexer.CreateBe } if err := i.db.InsertBeaconBadBlob(ctx, ProtoBeaconBadBlobToDBBeaconBadBlob(badBlob)); err != nil { + if persistence.IsUniqueConstraintError(err) { + return nil, status.Error(codes.AlreadyExists, "beacon blob already indexed") + } + i.log.WithError(err).WithFields(logFields).Error("Failed to index bad blob") return nil, status.Error(codes.Internal, "failed to index bad blob") @@ -1237,6 +1253,10 @@ func (i *Indexer) CreateExecutionBlockTrace(ctx context.Context, req *indexer.Cr } if err := i.db.InsertExecutionBlockTrace(ctx, ProtoExecutionBlockTraceToDBExecutionBlockTrace(trace)); err != nil { + if persistence.IsUniqueConstraintError(err) { + return nil, status.Error(codes.AlreadyExists, "execution block trace already indexed") + } + return nil, status.Error(codes.Internal, "failed to insert execution block trace") } @@ -1442,6 +1462,10 @@ func (i *Indexer) CreateExecutionBadBlock(ctx context.Context, req *indexer.Crea } if err := i.db.InsertExecutionBadBlock(ctx, ProtoExecutionBadBlockToDBExecutionBadBlock(block)); err != nil { + if persistence.IsUniqueConstraintError(err) { + return nil, status.Error(codes.AlreadyExists, "execution bad block already indexed") + } + return nil, status.Error(codes.Internal, "failed to insert execution bad block") } diff --git a/pkg/server/service/indexer/permanent_store.go b/pkg/server/service/indexer/permanent_store.go index e6d9814..bf107e9 100644 --- a/pkg/server/service/indexer/permanent_store.go +++ b/pkg/server/service/indexer/permanent_store.go @@ -368,12 +368,24 @@ func (p *PermanentStore) processBlock(ctx context.Context, block PermanentStoreB // recordPermanentBlock records the block in the PermanentBlock table. func (p *PermanentStore) recordPermanentBlock(ctx context.Context, block PermanentStoreBlock) error { // Record the block directly since we already checked earlier if it exists - return p.db.InsertPermanentBlock(ctx, &persistence.PermanentBlock{ + err := p.db.InsertPermanentBlock(ctx, &persistence.PermanentBlock{ //nolint:gosec // At the mercy of the database Slot: int64(block.Slot), BlockRoot: block.BlockRoot, Network: block.Network, }) + + // A unique constraint violation here means another caller already + // recorded this exact block -- for example two processBlock calls + // racing past the earlier existence checks while a slow copy is still + // in flight, which the distributed lock's fixed 30s TTL doesn't fully + // rule out. The row we wanted to exist now exists either way, so this + // is success, not failure. + if err != nil && persistence.IsUniqueConstraintError(err) { + return nil + } + + return err } // GetPermanentLocation returns the permanent location for a block. diff --git a/pkg/server/service/indexer/unique_constraint_test.go b/pkg/server/service/indexer/unique_constraint_test.go new file mode 100644 index 0000000..a50d813 --- /dev/null +++ b/pkg/server/service/indexer/unique_constraint_test.go @@ -0,0 +1,295 @@ +package indexer + +import ( + "context" + "fmt" + "os" + "sync" + "testing" + + pindexer "github.com/ethpandaops/tracoor/pkg/proto/tracoor/indexer" + "github.com/ethpandaops/tracoor/pkg/server/ethereum" + "github.com/ethpandaops/tracoor/pkg/server/persistence" + "github.com/ethpandaops/tracoor/pkg/store" + "github.com/sirupsen/logrus" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/timestamppb" + "google.golang.org/protobuf/types/known/wrapperspb" +) + +func newUniqueConstraintTestIndexer(t *testing.T) *Indexer { + t.Helper() + + ctx := context.Background() + + dbFile, err := os.CreateTemp("", "unique_constraint_app_*.db") + if err != nil { + t.Fatalf("failed to create temp db file: %v", err) + } + dbPath := dbFile.Name() + dbFile.Close() + os.Remove(dbPath) + + t.Cleanup(func() { + os.Remove(dbPath) + os.Remove(dbPath + "-wal") + os.Remove(dbPath + "-shm") + }) + + db, err := persistence.NewIndexer("unique-constraint-test", logrus.New(), persistence.Config{ + DSN: fmt.Sprintf("file:%s?parseTime=True", dbPath), + DriverName: "sqlite", + }, persistence.DefaultOptions().SetMetricsEnabled(false)) + if err != nil { + t.Fatalf("failed to create persistence indexer: %v", err) + } + if err := db.Start(ctx); err != nil { + t.Fatalf("failed to migrate: %v", err) + } + + basePath, err := os.MkdirTemp("", "unique_constraint_app_fs") + if err != nil { + t.Fatalf("failed to create temp fs dir: %v", err) + } + t.Cleanup(func() { os.RemoveAll(basePath) }) + + st, err := store.NewFSStore("unique-constraint-test", logrus.New(), &store.FSStoreConfig{BasePath: basePath}, &store.Options{}) + if err != nil { + t.Fatalf("failed to create FS store: %v", err) + } + + idx, err := NewIndexer(ctx, logrus.New(), &Config{}, db, st, ðereum.Config{}) + if err != nil { + t.Fatalf("failed to create indexer: %v", err) + } + + return idx +} + +// TestCreateBeaconState_ConcurrentIdenticalRequestsProduceExactlyOneRow is a +// regression test for the TOCTOU race the audit found: 40 concurrent, +// identical CreateBeaconState calls used to reliably produce 2-3 duplicate +// rows (confirmed empirically in the original audit). With a real database +// constraint backing the dedup check, exactly one must land, and every +// loser must get a clean AlreadyExists rather than a raw SQL error leaking +// through as an Internal error. +func TestCreateBeaconState_ConcurrentIdenticalRequestsProduceExactlyOneRow(t *testing.T) { + idx := newUniqueConstraintTestIndexer(t) + ctx := context.Background() + + location := "beacon_state/concurrent.ssz" + data := []byte("data") + + if _, err := idx.Store().SaveBeaconState(ctx, &store.SaveParams{Data: &data, Location: location}); err != nil { + t.Fatalf("failed to pre-upload blob: %v", err) + } + + makeReq := func() *pindexer.CreateBeaconStateRequest { + return &pindexer.CreateBeaconStateRequest{ + Node: wrapperspb.String("race-node"), + Slot: wrapperspb.UInt64(12345), + Epoch: wrapperspb.UInt64(1), + StateRoot: wrapperspb.String("0xconcurrent"), + FetchedAt: timestamppb.Now(), + BeaconImplementation: wrapperspb.String("teku"), + NodeVersion: wrapperspb.String("1.0.0"), + Location: wrapperspb.String(location), + Network: wrapperspb.String("mainnet"), + } + } + + const concurrency = 40 + + var ( + wg sync.WaitGroup + start = make(chan struct{}) + succeeded int + alreadyExist int + unexpected int + mu sync.Mutex + ) + + for i := 0; i < concurrency; i++ { + wg.Add(1) + + go func() { + defer wg.Done() + <-start + + _, err := idx.CreateBeaconState(ctx, makeReq()) + + mu.Lock() + defer mu.Unlock() + + switch { + case err == nil: + succeeded++ + case status.Code(err) == codes.AlreadyExists: + alreadyExist++ + default: + unexpected++ + t.Logf("unexpected error (not a clean AlreadyExists): %v", err) + } + }() + } + + close(start) + wg.Wait() + + countRsp, err := idx.CountBeaconState(ctx, &pindexer.CountBeaconStateRequest{}) + if err != nil { + t.Fatalf("failed to count: %v", err) + } + + t.Logf("concurrency=%d succeeded=%d alreadyExists=%d unexpected=%d dbRows=%d", + concurrency, succeeded, alreadyExist, unexpected, countRsp.Count.Value) + + if countRsp.Count.Value != 1 { + t.Fatalf("expected exactly 1 row from %d concurrent identical requests, got %d", concurrency, countRsp.Count.Value) + } + + if succeeded != 1 { + t.Fatalf("expected exactly 1 caller to see success, got %d", succeeded) + } + + if unexpected != 0 { + t.Fatalf("expected every losing caller to see a clean AlreadyExists, got %d unexpected error(s)", unexpected) + } + + if alreadyExist != concurrency-1 { + t.Fatalf("expected %d callers to see AlreadyExists, got %d", concurrency-1, alreadyExist) + } +} + +// TestCreateExecutionBadBlock_ConcurrentIdenticalRequestsProduceExactlyOneRow +// covers the handler that previously had zero protection of any kind (not +// even a racy check) -- confirmed in the original audit with a +// deterministic, non-concurrent PoC (2 sequential identical calls both +// succeeded). This exercises it under real concurrency too. +func TestCreateExecutionBadBlock_ConcurrentIdenticalRequestsProduceExactlyOneRow(t *testing.T) { + idx := newUniqueConstraintTestIndexer(t) + ctx := context.Background() + + makeReq := func() *pindexer.CreateExecutionBadBlockRequest { + return &pindexer.CreateExecutionBadBlockRequest{ + Node: wrapperspb.String("race-node"), + BlockHash: wrapperspb.String("0xconcurrent"), + BlockNumber: wrapperspb.Int64(1), + FetchedAt: timestamppb.Now(), + Location: wrapperspb.String("execution_bad_block/concurrent.json"), + ContentEncoding: wrapperspb.String("gzip"), + Network: wrapperspb.String("mainnet"), + ExecutionImplementation: wrapperspb.String("geth"), + NodeVersion: wrapperspb.String("1.0.0"), + } + } + + const concurrency = 40 + + var ( + wg sync.WaitGroup + start = make(chan struct{}) + succeeded int + alreadyExist int + unexpected int + mu sync.Mutex + ) + + for i := 0; i < concurrency; i++ { + wg.Add(1) + + go func() { + defer wg.Done() + <-start + + _, err := idx.CreateExecutionBadBlock(ctx, makeReq()) + + mu.Lock() + defer mu.Unlock() + + switch { + case err == nil: + succeeded++ + case status.Code(err) == codes.AlreadyExists: + alreadyExist++ + default: + unexpected++ + t.Logf("unexpected error (not a clean AlreadyExists): %v", err) + } + }() + } + + close(start) + wg.Wait() + + countRsp, err := idx.CountExecutionBadBlock(ctx, &pindexer.CountExecutionBadBlockRequest{}) + if err != nil { + t.Fatalf("failed to count: %v", err) + } + + t.Logf("concurrency=%d succeeded=%d alreadyExists=%d unexpected=%d dbRows=%d", + concurrency, succeeded, alreadyExist, unexpected, countRsp.Count.Value) + + if countRsp.Count.Value != 1 { + t.Fatalf("expected exactly 1 row from %d concurrent identical requests, got %d", concurrency, countRsp.Count.Value) + } + + if unexpected != 0 { + t.Fatalf("expected every losing caller to see a clean AlreadyExists, got %d unexpected error(s)", unexpected) + } +} + +// TestRecordPermanentBlock_ConcurrentCallsForSameIdentitySucceedExactlyOnce +// covers the NM-W2-001 interaction: two callers racing past the distributed +// lock for the same identity (which the lock's fixed 30s TTL with no +// renewal doesn't fully rule out) must both come away believing the block +// is durably recorded, not have one of them treat "someone else already +// recorded it" as a failure. +func TestRecordPermanentBlock_ConcurrentCallsForSameIdentitySucceedExactlyOnce(t *testing.T) { + idx := newUniqueConstraintTestIndexer(t) + ctx := context.Background() + + block := PermanentStoreBlock{ + Location: "beacon_block/permanent.ssz", + BlockRoot: "0xpermanentrace", + Network: "mainnet", + Slot: 1, + } + + const concurrency = 20 + + var ( + wg sync.WaitGroup + start = make(chan struct{}) + errs = make([]error, concurrency) + ) + + for i := 0; i < concurrency; i++ { + wg.Add(1) + + go func(idx2 int) { + defer wg.Done() + <-start + + errs[idx2] = idx.permanentStore.recordPermanentBlock(ctx, block) + }(i) + } + + close(start) + wg.Wait() + + for i, err := range errs { + if err != nil { + t.Fatalf("call %d: expected recordPermanentBlock to treat a concurrent duplicate as success, got: %v", i, err) + } + } + + permanentBlock, err := idx.db.GetPermanentBlockByBlockRoot(ctx, block.BlockRoot, block.Network) + if err != nil { + t.Fatalf("expected exactly one permanent block record to exist: %v", err) + } + if permanentBlock == nil { + t.Fatal("expected a non-nil permanent block record") + } +}