fix(db): drop the 9-column recommendations natural key index (#806)

internal/config/store_postgres_recommendations_test.go

@@ -606,7 +606,7 @@ func TestPostgresStore_UpsertRecommendations_AZArbiter(t *testing.T) {
 // tenancy updates in place, regional non-EC2 rows (platform and tenancy ”) are
 // unchanged, and a row written by pre-PR-C code is found and updated. A batch
 // of rows differing only by platform or tenancy is covered with the 9-column
-// index drop (migration 000112).
+// index drop (migration 000119).
 func TestPostgresStore_UpsertRecommendations_PlatformTenancyArbiter(t *testing.T) {
 	ctx := context.Background()
 	container, err := testhelpers.SetupPostgresContainer(ctx, t)
@@ -695,6 +695,9 @@ func TestPostgresStore_UpsertRecommendations_MultiAZBatch(t *testing.T) {
 	var oldIdx bool
 	require.NoError(t, pool.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM pg_indexes WHERE indexname = 'recommendations_natural_key_idx')`).Scan(&oldIdx))
 	require.False(t, oldIdx, "8-column index must be gone after the full migration chain")
+	var nineColIdx bool
+	require.NoError(t, pool.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM pg_indexes WHERE indexname = 'recommendations_natural_key_az_idx')`).Scan(&nineColIdx))
+	require.False(t, nineColIdx, "9-column index must be gone after the full migration chain (000119)")
 
 	savingsByAZ := func(t *testing.T) map[string]float64 {
 		t.Helper()
@@ -733,3 +736,82 @@ func TestPostgresStore_UpsertRecommendations_MultiAZBatch(t *testing.T) {
 		assert.Equal(t, map[string]float64{"us-east-1a": 1, "us-east-1b": 2}, savingsByAZ(t))
 	})
 }
+
+// Issue #806, PR-E: once migration 000119 drops the 9-column index, EC2 recs
+// that differ only by platform or tenancy upsert in ONE batch (the shape
+// persistCollection writes) without a unique violation, and both rows persist.
+func TestPostgresStore_UpsertRecommendations_PlatformTenancyBatch(t *testing.T) {
+	ctx := context.Background()
+	container, err := testhelpers.SetupPostgresContainer(ctx, t)
+	require.NoError(t, err)
+	defer container.Cleanup(ctx)
+	pool := container.DB.Pool()
+	require.NoError(t, migrations.RunMigrations(ctx, pool, getMigrationsPath(), "", ""))
+	store := config.NewPostgresStore(container.DB)
+	t0 := time.Now().UTC().Truncate(time.Second)
+	collect := []config.SuccessfulCollect{{Provider: "aws"}}
+
+	ec2 := func(platform, tenancy string, savings float64) config.RecommendationRecord {
+		rec := awsRec("p-"+platform+tenancy, "ec2", "us-east-1", "m5.large", savings)
+		rec.Details = json.RawMessage(`{"instance_type":"m5.large","platform":"` + platform + `","tenancy":"` + tenancy + `"}`)
+		return rec
+	}
+	savingsByKey := func(t *testing.T) map[string]float64 {
+		t.Helper()
+		rows, err := pool.Query(ctx, `SELECT platform || '|' || tenancy, monthly_savings FROM recommendations`)
+		require.NoError(t, err)
+		defer rows.Close()
+		out := map[string]float64{}
+		for rows.Next() {
+			var k string
+			var v float64
+			require.NoError(t, rows.Scan(&k, &v))
+			out[k] = v
+		}
+		require.NoError(t, rows.Err())
+		return out
+	}
+	reset := func(t *testing.T) {
+		t.Helper()
+		require.NoError(t, store.ReplaceRecommendations(ctx, t0, nil))
+	}
+
+	t.Run("two rows differing only by platform in one batch both persist", func(t *testing.T) {
+		reset(t)
+		require.NoError(t, store.UpsertRecommendations(ctx, t0,
+			[]config.RecommendationRecord{ec2("Linux/UNIX", "default", 10), ec2("Windows", "default", 20)}, collect))
+		assert.Equal(t, map[string]float64{"Linux/UNIX|default": 10, "Windows|default": 20}, savingsByKey(t))
+	})
+
+	t.Run("two rows differing only by tenancy in one batch both persist", func(t *testing.T) {
+		reset(t)
+		require.NoError(t, store.UpsertRecommendations(ctx, t0,
+			[]config.RecommendationRecord{ec2("Linux/UNIX", "default", 10), ec2("Linux/UNIX", "dedicated", 20)}, collect))
+		assert.Equal(t, map[string]float64{"Linux/UNIX|default": 10, "Linux/UNIX|dedicated": 20}, savingsByKey(t))
+	})
+
+	t.Run("same-key re-upsert updates in place and leaves the sibling alone", func(t *testing.T) {
+		require.NoError(t, store.UpsertRecommendations(ctx, t0.Add(time.Hour),
+			[]config.RecommendationRecord{ec2("Linux/UNIX", "default", 11), ec2("Linux/UNIX", "dedicated", 20)}, collect))
+		assert.Equal(t, map[string]float64{"Linux/UNIX|default": 11, "Linux/UNIX|dedicated": 20}, savingsByKey(t))
+	})
+
+	t.Run("regional non-EC2 row alongside EC2 platform rows is unaffected", func(t *testing.T) {
+		reset(t)
+		rds := awsRec("r1", "rds", "us-east-1", "db.m5.large", 30)
+		batch := []config.RecommendationRecord{ec2("Linux/UNIX", "default", 10), ec2("Windows", "default", 20), rds}
+		require.NoError(t, store.UpsertRecommendations(ctx, t0, batch, collect))
+		assert.Equal(t, map[string]float64{"Linux/UNIX|default": 10, "Windows|default": 20, "|": 30}, savingsByKey(t))
+	})
+
+	t.Run("cross-run Linux then Windows both persist", func(t *testing.T) {
+		// The transitional case that failed with SQLSTATE 23505 while the
+		// 9-column index was live: one collection stored Linux, a later one
+		// adds Windows on the same 9-column key (a successful collection sweeps
+		// rows it does not re-send, so the later batch carries both).
+		reset(t)
+		require.NoError(t, store.UpsertRecommendations(ctx, t0, []config.RecommendationRecord{ec2("Linux/UNIX", "default", 10)}, collect))
+		require.NoError(t, store.UpsertRecommendations(ctx, t0.Add(time.Hour), []config.RecommendationRecord{ec2("Linux/UNIX", "default", 10), ec2("Windows", "default", 20)}, collect))
+		assert.Equal(t, map[string]float64{"Linux/UNIX|default": 10, "Windows|default": 20}, savingsByKey(t))
+	})
+}

internal/database/postgres/migrations/000119_drop_recommendations_9col_natural_key.down.sql

@@ -0,0 +1,25 @@
+-- Rollback of 000119: recreate the 9-column unique index exactly as defined by
+-- 000109.
+--
+-- Only safe together with a code rollback to the pre-PR-C upsert (9-column
+-- ON CONFLICT). PR-C code needs the 11-column arbiter, so rolling PR-C back
+-- requires this down migration first, otherwise the old code runs without a
+-- 9-column arbiter. The recommendations table is a regenerable cache, so rows
+-- that differ only by platform or tenancy (allowed while the 9-column index is
+-- absent) are collapsed to the most recently collected one before the index is
+-- rebuilt; the next collection cycle repopulates the rest. Only duplicates
+-- under the 9-column key are deleted.
+DELETE FROM recommendations
+WHERE id IN (
+    SELECT id FROM (
+        SELECT id, ROW_NUMBER() OVER (
+            PARTITION BY account_key, provider, service, region, resource_type, engine, term, payment_option, availability_zone
+            ORDER BY collected_at DESC, id DESC
+        ) AS rn
+        FROM recommendations
+    ) ranked
+    WHERE rn > 1
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS recommendations_natural_key_az_idx
+    ON recommendations (account_key, provider, service, region, resource_type, engine, term, payment_option, availability_zone);

internal/database/postgres/migrations/000119_drop_recommendations_9col_natural_key.up.sql

@@ -0,0 +1,19 @@
+-- Migration 000119: drop the 9-column recommendations_natural_key_az_idx
+-- (#806, PR-E of the platform/tenancy identity work).
+--
+-- Migration 000111 added the generated platform and tenancy columns and the
+-- 11-column recommendations_natural_key_az_tp_idx, and kept the 9-column index
+-- so the pre-PR-C upsert (ON CONFLICT on the 9 columns) kept working. The store
+-- now upserts on the 11-column index (PR-C), so the 9-column index only does
+-- harm: it rejects two EC2 rows that differ only by platform or tenancy, which
+-- aborts the whole collection write. Dropping it lets those rows coexist.
+--
+-- DEPLOY ORDER: apply this only after the code that upserts on the 11-column
+-- index (PR-C) is live everywhere. Applying it while older code still runs
+-- breaks every collection, because ON CONFLICT on the 9 columns needs that
+-- index. database-migration.yml applies ALL pending migrations when `steps` is
+-- 0 or empty, so never dispatch it with the default before PR-C is released.
+--
+-- The 11-column index is strictly finer than the 9-column one, so no existing
+-- row can violate it and nothing needs backfilling.
+DROP INDEX IF EXISTS recommendations_natural_key_az_idx;

internal/database/postgres/migrations/000119_drop_recommendations_9col_natural_key_test.go

@@ -0,0 +1,114 @@
+//go:build integration
+// +build integration
+
+package migrations_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const insertRecForMigration119 = `
+	INSERT INTO recommendations
+	    (collected_at, provider, service, region, resource_type, engine, payload,
+	     upfront_cost, monthly_savings, term, payment_option)
+	VALUES ($1::timestamptz, 'aws', 'ec2', 'us-east-1', $2, '', $3::jsonb, 0, 0, 1, 'all-upfront')
+`
+
+// TestMigration_DropRecommendations9ColNaturalKey locks down migration 000119
+// (issue #806, PR-E): rows that differ only by platform or tenancy coexist once
+// the 9-column index is gone, the 11-column index still dedupes, and the
+// rollback recreates the 9-column index after collapsing those duplicates.
+func TestMigration_DropRecommendations9ColNaturalKey(t *testing.T) {
+	ctx := context.Background()
+	migrationsPath := getMigrationsPath()
+
+	container, err := testhelpers.SetupPostgresContainer(ctx, t)
+	require.NoError(t, err)
+	defer container.Cleanup(ctx)
+	pool := container.DB.Pool()
+
+	indexExists := func(name string) bool {
+		var exists bool
+		require.NoError(t, pool.QueryRow(ctx,
+			`SELECT EXISTS (SELECT 1 FROM pg_indexes WHERE tablename = 'recommendations' AND indexname = $1)`, name).Scan(&exists))
+		return exists
+	}
+	count := func(resourceType string) int {
+		var n int
+		require.NoError(t, pool.QueryRow(ctx, `SELECT COUNT(*) FROM recommendations WHERE resource_type = $1`, resourceType).Scan(&n))
+		return n
+	}
+	const (
+		linux   = `{"availability_zone":"us-east-1a","details":{"platform":"Linux/UNIX","tenancy":"default"}}`
+		windows = `{"availability_zone":"us-east-1a","details":{"platform":"Windows","tenancy":"default"}}`
+		rhel    = `{"availability_zone":"us-east-1a","details":{"platform":"Linux/UNIX","tenancy":"dedicated"}}`
+		other   = `{"availability_zone":"us-east-1b","details":{"platform":"Linux/UNIX","tenancy":"default"}}`
+	)
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 119)))
+	require.True(t, indexExists("recommendations_natural_key_az_idx"))
+	require.True(t, indexExists("recommendations_natural_key_az_tp_idx"))
+
+	// A regional row and one EC2 row exist before the migration.
+	_, err = pool.Exec(ctx, insertRecForMigration119, "2026-01-01T00:00:00Z", "regional", `{"details":{"scope":"Region"}}`)
+	require.NoError(t, err)
+	_, err = pool.Exec(ctx, insertRecForMigration119, "2026-01-01T00:00:00Z", "ec2", linux)
+	require.NoError(t, err)
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 119))
+	assert.False(t, indexExists("recommendations_natural_key_az_idx"), "9-column index must be dropped")
+	assert.True(t, indexExists("recommendations_natural_key_az_tp_idx"), "11-column index must remain")
+
+	// Rows differing only by platform or tenancy now coexist; an exact
+	// duplicate and a duplicate regional row are still rejected by the
+	// 11-column index.
+	_, err = pool.Exec(ctx, insertRecForMigration119, "2026-01-02T00:00:00Z", "ec2", windows)
+	require.NoError(t, err)
+	_, err = pool.Exec(ctx, insertRecForMigration119, "2026-01-03T00:00:00Z", "ec2", rhel)
+	require.NoError(t, err)
+	_, err = pool.Exec(ctx, insertRecForMigration119, "2026-01-03T00:00:00Z", "ec2", other)
+	require.NoError(t, err, "a different AZ is a different row under both indexes")
+	assert.Equal(t, 4, count("ec2"))
+	_, err = pool.Exec(ctx, insertRecForMigration119, "2026-01-03T00:00:00Z", "ec2", linux)
+	require.Error(t, err, "same platform and tenancy must still violate the 11-column index")
+	_, err = pool.Exec(ctx, insertRecForMigration119, "2026-01-03T00:00:00Z", "regional", `{"details":{"scope":"Region"}}`)
+	require.Error(t, err, "regional duplicate must still violate the 11-column index")
+
+	// Rollback: platform/tenancy-only duplicates collapse to the newest, the
+	// other-AZ row and the regional row are untouched, and the 9-column index
+	// is recreated with its original shape.
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 119)))
+	assert.Equal(t, 2, count("ec2"), "only duplicates under the 9-column key may be deleted")
+	assert.Equal(t, 1, count("regional"))
+	rows, err := pool.Query(ctx, `SELECT availability_zone, platform, tenancy FROM recommendations WHERE resource_type = 'ec2' ORDER BY availability_zone`)
+	require.NoError(t, err)
+	type kept struct{ az, platform, tenancy string }
+	var got []kept
+	for rows.Next() {
+		var k kept
+		require.NoError(t, rows.Scan(&k.az, &k.platform, &k.tenancy))
+		got = append(got, k)
+	}
+	require.NoError(t, rows.Err())
+	assert.Equal(t, []kept{
+		{"us-east-1a", "Linux/UNIX", "dedicated"}, // newest of the us-east-1a group
+		{"us-east-1b", "Linux/UNIX", "default"},
+	}, got)
+
+	var indexDef string
+	require.NoError(t, pool.QueryRow(ctx,
+		`SELECT indexdef FROM pg_indexes WHERE indexname = 'recommendations_natural_key_az_idx'`).Scan(&indexDef))
+	assert.Contains(t, indexDef, "UNIQUE INDEX")
+	assert.Contains(t, indexDef, "(account_key, provider, service, region, resource_type, engine, term, payment_option, availability_zone)")
+	assert.True(t, indexExists("recommendations_natural_key_az_tp_idx"))
+
+	// Re-applying the up migration after a rollback works.
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 119))
+	assert.False(t, indexExists("recommendations_natural_key_az_idx"))
+}

internal/scheduler/scheduler_global_filters_integration_test.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers"
@@ -147,3 +148,56 @@ func TestPersistCollectionMultiAZInOneBatch(t *testing.T) {
 	require.NoError(t, err)
 	assert.Nil(t, state.LastCollectionError, "persistCollection must not record a write error")
 }
+
+// Issue #806, PR-E: persistCollection writes every provider's recommendations
+// in one transaction, so two EC2 recs on the same 9-column key (differing only
+// by platform or tenancy) used to fail the whole collection write. The records
+// come from the real convertRecommendations, so the Details blob that feeds the
+// generated platform and tenancy columns is the one production writes.
+func TestPersistCollectionPlatformTenancyInOneBatch(t *testing.T) {
+	ctx := t.Context()
+	pg, err := testhelpers.SetupPostgresContainer(ctx, t)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, pg.Cleanup(context.Background())) })
+	require.NoError(t, migrations.RunMigrations(ctx, pg.DB.Pool(), "../database/postgres/migrations", "", ""))
+	store := config.NewPostgresStore(pg.DB)
+	s := &Scheduler{config: store}
+
+	ec2 := func(platform, tenancy string) common.Recommendation {
+		return common.Recommendation{
+			Provider: common.ProviderAWS, Service: common.ServiceEC2, Region: "us-east-1", ResourceType: "m5.large",
+			Count: 1, Term: "1yr", PaymentOption: "all-upfront", EstimatedSavings: 10,
+			Details: &common.ComputeDetails{InstanceType: "m5.large", Platform: platform, Tenancy: tenancy, Scope: "regional"},
+		}
+	}
+	persist := func(recs ...common.Recommendation) {
+		s.persistCollection(ctx, s.convertRecommendations(recs, "aws"), []config.SuccessfulCollect{{Provider: "aws"}}, nil)
+	}
+	keys := func() map[string]bool {
+		rows, err := pg.DB.Pool().Query(ctx, `SELECT platform || '|' || tenancy FROM recommendations`)
+		require.NoError(t, err)
+		defer rows.Close()
+		out := map[string]bool{}
+		for rows.Next() {
+			var k string
+			require.NoError(t, rows.Scan(&k))
+			out[k] = true
+		}
+		require.NoError(t, rows.Err())
+		return out
+	}
+	assertNoWriteError := func() {
+		state, err := store.GetRecommendationsFreshness(ctx)
+		require.NoError(t, err)
+		assert.Nil(t, state.LastCollectionError, "persistCollection must not record a write error")
+	}
+
+	persist(ec2("Linux/UNIX", "default"), ec2("Windows", "default"), ec2("Linux/UNIX", "dedicated"))
+	assert.Equal(t, map[string]bool{"Linux/UNIX|default": true, "Windows|default": true, "Linux/UNIX|dedicated": true}, keys())
+	assertNoWriteError()
+
+	// A later collection re-sends the same keys: updated in place, no growth.
+	persist(ec2("Linux/UNIX", "default"), ec2("Windows", "default"), ec2("Linux/UNIX", "dedicated"))
+	assert.Len(t, keys(), 3)
+	assertNoWriteError()
+}