From 3ca3721618349e13003d901939d452eecc15cce3 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Mon, 12 Oct 2026 22:28:41 +0200 Subject: [PATCH] fix(config): upsert recommendations on the 11-column index with platform and tenancy Point the recommendations upsert ON CONFLICT arbiter at recommendations_natural_key_az_tp_idx (the 9 existing key columns plus the generated platform and tenancy columns from migration 000111). The insert column list is unchanged and the generated columns are never written or updated. Add TestPostgresStore_UpsertRecommendations_PlatformTenancyArbiter (same-key re-upsert in place, regional non-EC2 rows with empty platform/tenancy, pre-PR-C row found). Give the legacy-row case in the AZ arbiter test the platform and tenancy the old collector always stored in details. Bump the Azure pricing integration test from migration 109 to 111, the only pinned-schema test that calls the recommendations store. Refs #806 Co-Authored-By: Claude Sonnet 5.5 --- .../config/store_postgres_recommendations.go | 9 ++- .../store_postgres_recommendations_test.go | 81 ++++++++++++++++++- .../azure_pricing_integration_test.go | 5 +- 3 files changed, 87 insertions(+), 8 deletions(-) diff --git a/internal/config/store_postgres_recommendations.go b/internal/config/store_postgres_recommendations.go index 1baaaf0e..f3d8b93b 100644 --- a/internal/config/store_postgres_recommendations.go +++ b/internal/config/store_postgres_recommendations.go @@ -63,7 +63,7 @@ func (s *PostgresStore) ReplaceRecommendations(ctx context.Context, collectedAt // UpsertRecommendations is the incremental write path: it upserts each row // by natural key (cloud_account_id, provider, service, region, -// resource_type, engine, term, payment_option, availability_zone) and then evicts stale rows for the +// resource_type, engine, term, payment_option, availability_zone, platform, tenancy) and then evicts stale rows for the // set of (provider, account) pairs that successfully collected in this run. // Pairs whose collection failed are NOT in successfulCollects and their // stale rows stay — callers see older data with a banner rather than a @@ -72,8 +72,9 @@ func (s *PostgresStore) ReplaceRecommendations(ctx context.Context, collectedAt // Migration 000032 broadened the natural key to include term + payment_option, // so per-rec ON CONFLICT no longer collides on SQLSTATE 21000. Migration // 000109 added availability_zone, a GENERATED column derived from payload that -// is never written here; the upsert arbiter is the 9-column -// recommendations_natural_key_az_idx (issue #792). +// is never written here. Migration 000111 added the platform and tenancy +// generated columns the same way; the upsert arbiter is the 11-column +// recommendations_natural_key_az_tp_idx (issues #792, #806). // // The eviction predicate uses (provider, account_key) IN (unnest($2, $3)) // where account_key matches the generated column on the table — nil @@ -227,7 +228,7 @@ func insertRecommendationsBatch(ctx context.Context, tx pgx.Tx, collectedAt time if onConflict { query += ` - ON CONFLICT (account_key, provider, service, region, resource_type, engine, term, payment_option, availability_zone) + ON CONFLICT (account_key, provider, service, region, resource_type, engine, term, payment_option, availability_zone, platform, tenancy) DO UPDATE SET payload = EXCLUDED.payload, upfront_cost = EXCLUDED.upfront_cost, diff --git a/internal/config/store_postgres_recommendations_test.go b/internal/config/store_postgres_recommendations_test.go index 9c4198a6..23367043 100644 --- a/internal/config/store_postgres_recommendations_test.go +++ b/internal/config/store_postgres_recommendations_test.go @@ -581,13 +581,14 @@ func TestPostgresStore_UpsertRecommendations_AZArbiter(t *testing.T) { t.Run("row written by pre-PR2b code is found and updated", func(t *testing.T) { require.NoError(t, store.ReplaceRecommendations(ctx, t0, nil)) - // What the old code stored: AZ only inside details, 8-column semantics. + // What the old code stored: AZ, platform and tenancy only inside details (the + // collector always sets all three on a zonal EC2 rec), 8-column semantics. _, err := pool.Exec(ctx, ` INSERT INTO recommendations (collected_at, provider, service, region, resource_type, engine, payload, upfront_cost, monthly_savings, term, payment_option) VALUES ($1, 'aws', 'ec2', 'us-east-1', 'm5.large', '', - '{"details":{"availability_zone":"us-east-1a"}}'::jsonb, 100, 5, 12, 'no-upfront')`, t0) + '{"details":{"platform":"Linux/UNIX","tenancy":"default","availability_zone":"us-east-1a"}}'::jsonb, 100, 5, 12, 'no-upfront')`, t0) require.NoError(t, err) require.NoError(t, store.UpsertRecommendations(ctx, t0.Add(time.Hour), []config.RecommendationRecord{zonalAWSRec("us-east-1a", 40)}, collect)) var n int @@ -600,6 +601,82 @@ func TestPostgresStore_UpsertRecommendations_AZArbiter(t *testing.T) { }) } +// Issue #806, PR-C: the upsert arbiter is the 11-column index that adds the +// generated platform and tenancy columns. A re-upsert on the same platform and +// 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). +func TestPostgresStore_UpsertRecommendations_PlatformTenancyArbiter(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 + } + columns := func(t *testing.T) (platform, tenancy string, n int) { + t.Helper() + require.NoError(t, pool.QueryRow(ctx, `SELECT COUNT(*) FROM recommendations`).Scan(&n)) + require.NoError(t, pool.QueryRow(ctx, `SELECT platform, tenancy FROM recommendations LIMIT 1`).Scan(&platform, &tenancy)) + return platform, tenancy, n + } + + t.Run("same platform and tenancy re-upsert updates in place", func(t *testing.T) { + require.NoError(t, store.UpsertRecommendations(ctx, t0, []config.RecommendationRecord{ec2("Windows", "dedicated", 10)}, collect)) + require.NoError(t, store.UpsertRecommendations(ctx, t0.Add(time.Hour), []config.RecommendationRecord{ec2("Windows", "dedicated", 25)}, collect)) + got, err := store.ListStoredRecommendations(ctx, config.RecommendationFilter{}) + require.NoError(t, err) + require.Len(t, got, 1) + assert.InDelta(t, 25.0, got[0].Savings, 0.001) + platform, tenancy, n := columns(t) + assert.Equal(t, 1, n) + assert.Equal(t, "Windows", platform, "the generated column is derived from the payload") + assert.Equal(t, "dedicated", tenancy) + }) + + t.Run("regional non-EC2 row has empty platform and tenancy and updates in place", func(t *testing.T) { + require.NoError(t, store.ReplaceRecommendations(ctx, t0, nil)) + rds := awsRec("r1", "rds", "us-east-1", "db.m5.large", 10) + require.NoError(t, store.UpsertRecommendations(ctx, t0, []config.RecommendationRecord{rds}, collect)) + rds.Savings = 30 + require.NoError(t, store.UpsertRecommendations(ctx, t0.Add(time.Hour), []config.RecommendationRecord{rds}, collect)) + got, err := store.ListStoredRecommendations(ctx, config.RecommendationFilter{}) + require.NoError(t, err) + require.Len(t, got, 1) + assert.InDelta(t, 30.0, got[0].Savings, 0.001) + platform, tenancy, n := columns(t) + assert.Equal(t, 1, n) + assert.Empty(t, platform) + assert.Empty(t, tenancy) + }) + + t.Run("row written by pre-PR-C code is found and updated", func(t *testing.T) { + require.NoError(t, store.ReplaceRecommendations(ctx, t0, nil)) + // What the old code stored: platform and tenancy only inside details. + _, err := pool.Exec(ctx, ` + INSERT INTO recommendations + (collected_at, provider, service, region, resource_type, engine, payload, + upfront_cost, monthly_savings, term, payment_option) + VALUES ($1, 'aws', 'ec2', 'us-east-1', 'm5.large', '', + '{"details":{"platform":"Linux/UNIX","tenancy":"default"}}'::jsonb, 100, 5, 12, 'no-upfront')`, t0) + require.NoError(t, err) + require.NoError(t, store.UpsertRecommendations(ctx, t0.Add(time.Hour), []config.RecommendationRecord{ec2("Linux/UNIX", "default", 40)}, collect)) + got, err := store.ListStoredRecommendations(ctx, config.RecommendationFilter{}) + require.NoError(t, err) + require.Len(t, got, 1, "the legacy row must be the conflict target, not duplicated") + 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 diff --git a/internal/purchase/azure_pricing_integration_test.go b/internal/purchase/azure_pricing_integration_test.go index 85d98d65..93dae37a 100644 --- a/internal/purchase/azure_pricing_integration_test.go +++ b/internal/purchase/azure_pricing_integration_test.go @@ -102,8 +102,9 @@ func TestApproveAzurePricingRequiresMigrationAndRecollection(t *testing.T) { require.NoError(t, migrations.MigrateToVersion(ctx, pool, rampMigrationsPath(), 106)) t.Run("106 invalidated old prices", func(t *testing.T) { approve(t, ErrStaleAzurePricing, 0) }) rec.UpfrontCost = 700 - // The store's upsert needs the availability_zone column from migration 000109. - require.NoError(t, migrations.MigrateToVersion(ctx, pool, rampMigrationsPath(), 109)) + // The store's upsert needs the availability_zone, platform and tenancy columns and the + // 11-column index from migration 000111. + require.NoError(t, migrations.MigrateToVersion(ctx, pool, rampMigrationsPath(), 111)) require.NoError(t, pricing.UpsertRecommendations(ctx, time.Now().Add(time.Minute), []config.RecommendationRecord{rec}, collects)) t.Run("recollected prices purchase once", func(t *testing.T) { approve(t, nil, 1) }) }