tapdb: dedupe legacy duplicate events during migration 62 backfill

The bug this PR closes -- restart-triggered re-fires of mint events
inserting duplicate supply_update_events rows -- could have already
fired against an unupgraded database. Two rows with identical content
hash to the same event_key, so the second SetSupplyUpdateEventKey in
the migration 62 backfill would violate the unique index added in
migration 61 and abort the migration.

Track hashes seen during the backfill loop and delete any row whose
hash a prior row already claimed. The deleted rows are by definition
the same logical event as the one that survives. A new
DeleteSupplyUpdateEvent sqlc query supports the delete-by-event_id
that the dedupe needs.

TestMigration62BackfillDedupesLegacyDuplicates seeds three identical
rows pre-backfill and asserts the migration leaves exactly one with
the expected hash, never erroring out.

Author: jtobin

tapdb/migrations_test.go

@@ -1,6 +1,7 @@
 package tapdb
 
 import (
+	"bytes"
 	"context"
 	"database/sql"
 	"encoding/hex"
@@ -1491,3 +1492,60 @@ func TestMigration62BackfillSupplyUpdateEventKeys(t *testing.T) {
 	require.NoError(t, postRows.Close())
 	require.Equal(t, len(seeds), seen)
 }
+
+// TestMigration62BackfillDedupesLegacyDuplicates simulates the legacy
+// failure mode this PR closes: pre-migration databases could contain
+// multiple supply_update_events rows with identical content. The
+// migration 62 backfill must drop the duplicates rather than fail on
+// the unique index added in migration 61.
+func TestMigration62BackfillDedupesLegacyDuplicates(t *testing.T) {
+	ctx := context.Background()
+
+	db := NewTestDBWithVersion(t, 61)
+
+	groupKey := bytes.Repeat([]byte{0x42}, 32)
+	payload := []byte("event-payload-duplicate")
+
+	// Insert the same logical event three times. NULL event_key is
+	// distinct from NULL under both backends, so all three rows
+	// land without tripping the unique index.
+	for i := 0; i < 3; i++ {
+		_, err := db.InsertSupplyUpdateEvent(
+			ctx, sqlc.InsertSupplyUpdateEventParams{
+				GroupKey:     groupKey,
+				TransitionID: sql.NullInt64{},
+				UpdateTypeID: 0,
+				EventData:    payload,
+				EventKey:     nil,
+			},
+		)
+		require.NoError(t, err)
+	}
+
+	var preCount int
+	require.NoError(t, db.QueryRowContext(ctx, `
+		SELECT COUNT(*) FROM supply_update_events
+	`).Scan(&preCount))
+	require.Equal(t, 3, preCount)
+
+	// Run the backfill. The unique index added in migration 61
+	// would reject the naive UPDATE for the second and third
+	// rows; the backfill must dedupe before writing.
+	err := db.ExecuteMigrations(TargetLatest, WithProgrammaticMigrations(
+		makeProgrammaticMigrations(db, programmaticMigrations, true),
+	))
+	require.NoError(t, err)
+
+	// Exactly one row should survive, and its event_key should
+	// match the hash of the duplicated content.
+	var postCount int
+	var survivingKey []byte
+	require.NoError(t, db.QueryRowContext(ctx, `
+		SELECT COUNT(*), MAX(event_key) FROM supply_update_events
+	`).Scan(&postCount, &survivingKey))
+	require.Equal(t, 1, postCount)
+	require.Equal(t,
+		supplyUpdateEventKey(groupKey, 0, payload),
+		survivingKey,
+	)
+}

tapdb/programmatic_migrations.go

@@ -342,6 +342,13 @@ func insertAssetBurns(ctx context.Context, q sqlc.Querier) error {
 // in migration 000061) and stores it in the new column. After this
 // migration runs every row holds a hash, and the unique index on
 // event_key enforces the no-duplicates invariant for new inserts.
+//
+// Legacy databases may already contain duplicate rows (the bug this
+// PR fixes -- restart re-fires of the same logical event). Two rows
+// with identical content hash to the same key, so the second
+// SetSupplyUpdateEventKey would violate the unique index added in
+// migration 000061. We dedupe in-memory by tracking the hashes we've
+// already assigned and dropping any row whose hash we've seen.
 func backfillSupplyUpdateEventKeys(ctx context.Context,
 	q sqlc.Querier) error {
 
@@ -354,11 +361,32 @@ func backfillSupplyUpdateEventKeys(ctx context.Context,
 	log.Debugf("Backfilling event_key for %d supply update events",
 		len(rows))
 
+	seen := make(map[string]struct{}, len(rows))
 	for _, row := range rows {
 		key := supplyUpdateEventKey(
 			row.GroupKey, row.UpdateTypeID, row.EventData,
 		)
 
+		if _, dup := seen[string(key)]; dup {
+			// A prior row in this loop already claimed this
+			// hash, so the current row is a duplicate of an
+			// earlier logical event. Drop it; the unique
+			// index in migration 000061 would otherwise
+			// reject the UPDATE below.
+			log.Debugf("Dropping duplicate supply update "+
+				"event %d during backfill", row.EventID)
+
+			err := q.DeleteSupplyUpdateEvent(ctx, row.EventID)
+			if err != nil {
+				return fmt.Errorf("error deleting "+
+					"duplicate event %d: %w",
+					row.EventID, err)
+			}
+
+			continue
+		}
+		seen[string(key)] = struct{}{}
+
 		err := q.SetSupplyUpdateEventKey(
 			ctx, sqlc.SetSupplyUpdateEventKeyParams{
 				EventKey: key,

tapdb/sqlc/querier.go

@@ -233,6 +233,13 @@ WHERE transition_id = @transition_id;
 DELETE FROM supply_update_events
 WHERE transition_id = @transition_id;
 
+-- name: DeleteSupplyUpdateEvent :exec
+-- Deletes a single supply update event row identified by its
+-- event_id. Used by the migration 62 backfill to drop duplicate
+-- rows that hash to the same event_key as an earlier row.
+DELETE FROM supply_update_events
+WHERE event_id = @event_id;
+
 -- name: FetchUnspentSupplyPreCommits :many
 -- Fetch unspent supply pre-commitment outputs. Each pre-commitment output
 -- comes from a mint anchor transaction and relates to an asset issuance