fix(db): drop the 8-column recommendations unique index so zonal EC2 AZ rows coexist (#792)

internal/config/store_postgres_recommendations_test.go

@@ -542,8 +542,8 @@ func zonalAWSRec(az string, savings float64) config.RecommendationRecord {
 // generated availability_zone column. Same-AZ re-upserts update in place, a
 // regional row (AZ ”) behaves as before, and a row written by pre-PR2b code
 // (no top-level availability_zone in the payload) is found and updated.
-// Multi-AZ rows on one 8-tuple stay rejected until the 8-column index is
-// dropped (PR2c), so they are not asserted here.
+// Multi-AZ rows on one 8-tuple are covered by
+// TestPostgresStore_UpsertRecommendations_MultiAZBatch (PR2c).
 func TestPostgresStore_UpsertRecommendations_AZArbiter(t *testing.T) {
 	ctx := context.Background()
 	container, err := testhelpers.SetupPostgresContainer(ctx, t)
@@ -599,3 +599,60 @@ func TestPostgresStore_UpsertRecommendations_AZArbiter(t *testing.T) {
 		assert.InDelta(t, 40.0, got[0].Savings, 0.001)
 	})
 }
+
+// Issue #792, PR2c: once migration 000110 drops the 8-column index, two zonal
+// EC2 recommendations that differ only by availability zone upsert in ONE batch
+// (the shape persistCollection writes) without "ON CONFLICT DO UPDATE command
+// cannot affect row a second time", and both rows persist.
+func TestPostgresStore_UpsertRecommendations_MultiAZBatch(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"}}
+
+	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")
+
+	savingsByAZ := func(t *testing.T) map[string]float64 {
+		t.Helper()
+		got, err := store.ListStoredRecommendations(ctx, config.RecommendationFilter{})
+		require.NoError(t, err)
+		out := map[string]float64{}
+		for _, r := range got {
+			out[r.AvailabilityZone] = r.Savings
+		}
+		require.Len(t, out, len(got), "no two stored rows may share an AZ here")
+		return out
+	}
+
+	t.Run("two zonal rows in one batch both persist", func(t *testing.T) {
+		batch := []config.RecommendationRecord{zonalAWSRec("us-east-1a", 10), zonalAWSRec("us-east-1b", 20)}
+		require.NoError(t, store.UpsertRecommendations(ctx, t0, batch, collect))
+		assert.Equal(t, map[string]float64{"us-east-1a": 10, "us-east-1b": 20}, savingsByAZ(t))
+	})
+
+	t.Run("same-AZ 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{zonalAWSRec("us-east-1a", 11), zonalAWSRec("us-east-1b", 20)}, collect))
+		assert.Equal(t, map[string]float64{"us-east-1a": 11, "us-east-1b": 20}, savingsByAZ(t))
+	})
+
+	t.Run("regional row alongside zonal rows is unaffected", func(t *testing.T) {
+		regional := awsRec("r1", "ec2", "us-east-1", "m5.large", 30)
+		batch := []config.RecommendationRecord{zonalAWSRec("us-east-1a", 12), zonalAWSRec("us-east-1b", 21), regional}
+		require.NoError(t, store.UpsertRecommendations(ctx, t0.Add(2*time.Hour), batch, collect))
+		assert.Equal(t, map[string]float64{"us-east-1a": 12, "us-east-1b": 21, "": 30}, savingsByAZ(t))
+	})
+
+	t.Run("ReplaceRecommendations stores the same AZ pair", func(t *testing.T) {
+		batch := []config.RecommendationRecord{zonalAWSRec("us-east-1a", 1), zonalAWSRec("us-east-1b", 2)}
+		require.NoError(t, store.ReplaceRecommendations(ctx, t0.Add(3*time.Hour), batch))
+		assert.Equal(t, map[string]float64{"us-east-1a": 1, "us-east-1b": 2}, savingsByAZ(t))
+	})
+}

internal/database/postgres/migrations/000110_drop_recommendations_8col_natural_key.down.sql

@@ -0,0 +1,22 @@
+-- Rollback of 000110: recreate the 8-column unique index exactly as defined by
+-- 000043.
+--
+-- Only safe together with a code rollback to the pre-PR2c upsert (8-column
+-- ON CONFLICT). The recommendations table is a regenerable cache, so rows that
+-- differ only by availability zone (allowed while the 8-column index is absent)
+-- are collapsed to the most recently collected one before the index is rebuilt;
+-- the next collection cycle repopulates the rest.
+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
+            ORDER BY collected_at DESC, id DESC
+        ) AS rn
+        FROM recommendations
+    ) ranked
+    WHERE rn > 1
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS recommendations_natural_key_idx
+    ON recommendations (account_key, provider, service, region, resource_type, engine, term, payment_option);

internal/database/postgres/migrations/000110_drop_recommendations_8col_natural_key.up.sql

@@ -0,0 +1,17 @@
+-- Migration 000110: drop the 8-column recommendations_natural_key_idx
+-- (#792, PR2c of the zonal EC2 AZ work).
+--
+-- Migration 000109 added the generated availability_zone column and the
+-- 9-column recommendations_natural_key_az_idx, and kept the 8-column index so
+-- the pre-PR2b upsert (ON CONFLICT on the 8 columns) kept working. The store now
+-- upserts on the 9-column index (PR2b), so the 8-column index only does harm:
+-- it rejects two zonal EC2 rows that differ only by availability zone, which
+-- aborts the whole collection write. Dropping it lets those rows coexist.
+--
+-- DEPLOY ORDER: apply this only after the code that upserts on the 9-column
+-- index (PR2b) is live. Applying it while the old code still runs breaks every
+-- collection, because ON CONFLICT on the 8 columns needs that index.
+--
+-- The 9-column index is strictly finer than the 8-column one, so no existing
+-- row can violate it and nothing needs backfilling.
+DROP INDEX IF EXISTS recommendations_natural_key_idx;

internal/database/postgres/migrations/000110_drop_recommendations_8col_natural_key_test.go

@@ -0,0 +1,92 @@
+//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 insertRecForMigration110 = `
+	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_DropRecommendations8ColNaturalKey locks down migration 000110
+// (issue #792, PR2c): rows that differ only by availability zone coexist once
+// the 8-column index is gone, the 9-column index still dedupes per AZ, and the
+// rollback recreates the 8-column index after collapsing AZ-only duplicates.
+func TestMigration_DropRecommendations8ColNaturalKey(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
+	}
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 109))
+	require.True(t, indexExists("recommendations_natural_key_idx"))
+
+	// A regional row and one zonal row exist before the migration.
+	_, err = pool.Exec(ctx, insertRecForMigration110, "2026-01-01T00:00:00Z", "regional", `{"details":{"scope":"Region"}}`)
+	require.NoError(t, err)
+	_, err = pool.Exec(ctx, insertRecForMigration110, "2026-01-01T00:00:00Z", "zonal", `{"availability_zone":"us-east-1a"}`)
+	require.NoError(t, err)
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 110))
+	assert.False(t, indexExists("recommendations_natural_key_idx"), "8-column index must be dropped")
+	assert.True(t, indexExists("recommendations_natural_key_az_idx"), "9-column index must remain")
+
+	// Two zonal rows differing only by AZ now coexist; a same-AZ duplicate and a
+	// duplicate regional row are still rejected by the 9-column index.
+	_, err = pool.Exec(ctx, insertRecForMigration110, "2026-01-02T00:00:00Z", "zonal", `{"availability_zone":"us-east-1b"}`)
+	require.NoError(t, err)
+	_, err = pool.Exec(ctx, insertRecForMigration110, "2026-01-03T00:00:00Z", "zonal", `{"availability_zone":"us-east-1c"}`)
+	require.NoError(t, err)
+	assert.Equal(t, 3, count("zonal"))
+	_, err = pool.Exec(ctx, insertRecForMigration110, "2026-01-03T00:00:00Z", "zonal", `{"availability_zone":"us-east-1a"}`)
+	require.Error(t, err, "same-AZ duplicate must still violate the 9-column index")
+	_, err = pool.Exec(ctx, insertRecForMigration110, "2026-01-03T00:00:00Z", "regional", `{"details":{"scope":"Region"}}`)
+	require.Error(t, err, "regional duplicate must still violate the 9-column index")
+
+	// Rollback: AZ-only duplicates collapse to the newest, the regional row is
+	// untouched, and the 8-column index is recreated with its original shape.
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 109))
+	assert.Equal(t, 1, count("zonal"), "AZ-only duplicates must be collapsed so the 8-column index can be rebuilt")
+	assert.Equal(t, 1, count("regional"))
+	var az string
+	require.NoError(t, pool.QueryRow(ctx, `SELECT availability_zone FROM recommendations WHERE resource_type = 'zonal'`).Scan(&az))
+	assert.Equal(t, "us-east-1c", az, "the most recently collected row must survive")
+
+	var indexDef string
+	require.NoError(t, pool.QueryRow(ctx,
+		`SELECT indexdef FROM pg_indexes WHERE indexname = 'recommendations_natural_key_idx'`).Scan(&indexDef))
+	assert.Contains(t, indexDef, "UNIQUE INDEX")
+	assert.Contains(t, indexDef, "(account_key, provider, service, region, resource_type, engine, term, payment_option)")
+	assert.True(t, indexExists("recommendations_natural_key_az_idx"))
+
+	// Re-applying the up migration after a rollback works.
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 110))
+	assert.False(t, indexExists("recommendations_natural_key_idx"))
+}

internal/scheduler/scheduler_global_filters_integration_test.go

@@ -4,6 +4,7 @@ package scheduler
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"testing"
 	"time"
@@ -100,3 +101,49 @@ func TestGlobalFiltersPersistedAmbientAndRegistered(t *testing.T) {
 		assert.ElementsMatch(t, []string{"0-other", "1-other", "1-1", "1-10", "1-11", "1-sibling"}, ids)
 	})
 }
+
+// Issue #792, PR2c: persistCollection writes every provider's recommendations
+// in one transaction, so two zonal EC2 recs on the same 8-tuple (differing only
+// by AZ) used to fail the whole collection write. With the 8-column index gone
+// the whole 500-row batch lands, siblings included.
+func TestPersistCollectionMultiAZInOneBatch(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)
+
+	zonal := func(az string) config.RecommendationRecord {
+		return config.RecommendationRecord{
+			ID: "ec2-" + az, Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.large",
+			AvailabilityZone: az, Count: 1, Term: 1, Payment: "all-upfront", Savings: 10,
+			Details: json.RawMessage(`{"instance_type":"m5.large","scope":"Availability Zone","availability_zone":"` + az + `"}`),
+		}
+	}
+	const total = 500
+	recs := []config.RecommendationRecord{zonal("us-east-1a"), zonal("us-east-1b")}
+	for i := len(recs); i < total; i++ {
+		recs = append(recs, config.RecommendationRecord{
+			ID: fmt.Sprintf("filler-%d", i), Provider: "aws", Service: "ec2", Region: "us-west-2",
+			ResourceType: fmt.Sprintf("c5.type%d", i), Count: 1, Term: 1, Payment: "all-upfront", Savings: 1,
+		})
+	}
+
+	s := &Scheduler{config: store}
+	s.persistCollection(ctx, recs, []config.SuccessfulCollect{{Provider: "aws"}}, nil)
+
+	rows, err := store.ListStoredRecommendations(ctx, config.RecommendationFilter{})
+	require.NoError(t, err)
+	require.Len(t, rows, total, "the whole collection must persist, not fail on the AZ pair")
+	azs := map[string]bool{}
+	for _, r := range rows {
+		if r.ResourceType == "m5.large" {
+			azs[r.AvailabilityZone] = true
+		}
+	}
+	assert.Equal(t, map[string]bool{"us-east-1a": true, "us-east-1b": true}, azs)
+	state, err := store.GetRecommendationsFreshness(ctx)
+	require.NoError(t, err)
+	assert.Nil(t, state.LastCollectionError, "persistCollection must not record a write error")
+}