fix(recommendations): key stored recs, rec ID and purchase identity on RDS az_config (#408)

internal/api/az_config_gate.go

@@ -0,0 +1,35 @@
+package api
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+)
+
+// azConfigSchemaChecker is implemented by the Postgres store. Stores without a
+// schema (test doubles) have nothing to check.
+type azConfigSchemaChecker interface {
+	CheckAZConfigMigration(ctx context.Context) error
+}
+
+// requireAZConfigSchema refuses a purchase submission before its transaction
+// opens when the az_config expand migrations (000154, 000155) are missing: the
+// suppression insert names az_config, so it would fail mid-transaction.
+func (h *Handler) requireAZConfigSchema(ctx context.Context) error {
+	checker, ok := h.config.(azConfigSchemaChecker)
+	if !ok {
+		return nil
+	}
+	err := checker.CheckAZConfigMigration(ctx)
+	if errors.Is(err, config.ErrAZConfigMigrationNotReady) {
+		return NewClientErrorWithDetails(503,
+			fmt.Sprintf("purchases are unavailable until database migrations 000154 and 000155 are applied: %v", err),
+			map[string]any{})
+	}
+	if err != nil {
+		return fmt.Errorf("check az_config migration: %w", err)
+	}
+	return nil
+}

internal/api/azconfig_expand_order_integration_test.go

@@ -0,0 +1,113 @@
+//go:build integration
+
+package api
+
+import (
+	"context"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"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"
+)
+
+// contractMigrationVersion is the version of the migration that drops the
+// 11-column recommendations key (#408). It can only run after the release that
+// upserts on the 12-column key, so every expand step the new code needs must
+// sort BEFORE it.
+func contractMigrationVersion(t *testing.T) uint {
+	t.Helper()
+	paths, err := filepath.Glob(filepath.Join(getMigrationsPath(), "*_drop_recommendations_11col_natural_key.up.sql"))
+	require.NoError(t, err)
+	require.Len(t, paths, 1)
+	n, err := strconv.ParseUint(strings.SplitN(filepath.Base(paths[0]), "_", 2)[0], 10, 64)
+	require.NoError(t, err)
+	return uint(n)
+}
+
+// previousVersion returns the highest migration version below version, so the
+// test stops at the last migration a deploy applies before the contract step.
+func previousVersion(t *testing.T, version uint) uint {
+	t.Helper()
+	paths, err := filepath.Glob(filepath.Join(getMigrationsPath(), "*.up.sql"))
+	require.NoError(t, err)
+	var best uint
+	for _, p := range paths {
+		n, err := strconv.ParseUint(strings.SplitN(filepath.Base(p), "_", 2)[0], 10, 64)
+		require.NoError(t, err)
+		if uint(n) < version && uint(n) > best {
+			best = uint(n)
+		}
+	}
+	require.NotZero(t, best)
+	return best
+}
+
+func expandOrderRDSRec(azConfig string, upfront float64) config.RecommendationRecord {
+	rec := rdsAZConfigRec(azConfig, upfront)
+	rec.ID = ""
+	return rec
+}
+
+// exerciseNewCode runs the four paths the #408 code touches against whatever
+// schema the database currently has.
+// Before the contract step the 11-column index is still in place, so a
+// Single-AZ and Multi-AZ pair of one class cannot coexist yet (the documented
+// limitation the contract step resolves, and the same failure the previous
+// release had); only after it may both variants be upserted in one batch.
+func exerciseNewCode(ctx context.Context, t *testing.T, store *config.PostgresStore, recs []config.RecommendationRecord, priced string, wantUpfront float64) {
+	t.Helper()
+	single := expandOrderRDSRec("single-az", 100)
+
+	require.NoError(t, store.UpsertRecommendations(ctx, time.Now(), recs, []config.SuccessfulCollect{{Provider: "aws"}}),
+		"upsert of RDS variants")
+
+	purchase := single
+	purchase.Count = 2
+	sups := config.BuildSuppressions([]config.RecommendationRecord{purchase}, uuid.NewString(), &config.GlobalConfig{}, time.Now())
+	require.Len(t, sups, 1)
+	require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { return store.CreateSuppressionTx(ctx, tx, &sups[0]) }),
+		"suppression write inside the purchase transaction")
+
+	active, err := store.ListActiveSuppressions(ctx)
+	require.NoError(t, err, "ListActiveSuppressions")
+	assert.NotEmpty(t, active)
+
+	h := &Handler{config: store}
+	req := []config.RecommendationRecord{{
+		Provider: "aws", Service: "rds", Region: "us-east-1", ResourceType: "db.r5.large", Engine: "mysql",
+		Term: 1, Payment: "all-upfront", Count: 1, Details: rdsDetails(priced),
+	}}
+	require.NoError(t, h.priceRecommendationsFromStore(ctx, req), "purchase pricing index lookup")
+	assert.Equal(t, wantUpfront, req[0].UpfrontCost)
+}
+
+// Issue #408 deploy order: the new code ships after the expand migrations and
+// before the contract one. A database stopped just below the contract step
+// must already serve every new-code path, and still do so once the contract
+// step has run.
+func TestAZConfigNewCodeWorksBeforeAndAfterContractMigration(t *testing.T) {
+	ctx := context.Background()
+	pg := testhelpers.RequireEmptyDatabase(ctx, t)
+	pool, path := pg.DB.Pool(), getMigrationsPath()
+	contract := contractMigrationVersion(t)
+	store := config.NewPostgresStore(pg.DB)
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, previousVersion(t, contract)))
+	single, multi := expandOrderRDSRec("single-az", 100), expandOrderRDSRec("multi-az", 300)
+	exerciseNewCode(ctx, t, store, []config.RecommendationRecord{single}, "single-az", 100)
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, contract))
+	_, err := pool.Exec(ctx, `TRUNCATE recommendations, purchase_suppressions`)
+	require.NoError(t, err)
+	exerciseNewCode(ctx, t, store, []config.RecommendationRecord{single, multi}, "multi-az", 300)
+}

internal/api/azconfig_gate_integration_test.go

@@ -0,0 +1,75 @@
+//go:build integration
+
+package api
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"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"
+)
+
+// txCountingStore counts the transactions the purchase path opens.
+type txCountingStore struct {
+	*config.PostgresStore
+	txs int
+}
+
+func (s *txCountingStore) WithTx(ctx context.Context, fn func(pgx.Tx) error) error {
+	s.txs++
+	return s.PostgresStore.WithTx(ctx, fn)
+}
+
+// Without the az_config expand migrations the new code must refuse loudly:
+// the purchase path before it opens a transaction, collection and suppression
+// listing with the typed error, not a SQL failure.
+func TestAZConfigGate_RefusesBeforeExpandMigrations(t *testing.T) {
+	ctx := context.Background()
+	pg := testhelpers.RequireEmptyDatabase(ctx, t)
+	pool, path := pg.DB.Pool(), getMigrationsPath()
+	store := &txCountingStore{PostgresStore: config.NewPostgresStore(pg.DB)}
+	h := &Handler{config: store}
+	exec := &config.PurchaseExecution{ExecutionID: uuid.NewString()}
+
+	for _, version := range []uint{153, 154} { // before both expands, then after only the first
+		require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, version))
+
+		_, err := h.persistExecutionAndSuppressions(ctx, exec, nil, "creator", "key")
+		ce, ok := IsClientError(err)
+		require.True(t, ok, "version %d: %v", version, err)
+		assert.Equal(t, 503, ce.code)
+		assert.Contains(t, ce.message, "000154 and 000155")
+		assert.Zero(t, store.txs, "no transaction may open before the gate")
+
+		rec := expandOrderRDSRec("single-az", 100)
+		err = store.UpsertRecommendations(ctx, time.Now(), []config.RecommendationRecord{rec}, []config.SuccessfulCollect{{Provider: "aws"}})
+		assert.True(t, errors.Is(err, config.ErrAZConfigMigrationNotReady), "collection: %v", err)
+
+		_, err = store.ListActiveSuppressions(ctx)
+		assert.True(t, errors.Is(err, config.ErrAZConfigMigrationNotReady), "suppression listing: %v", err)
+
+		sup := &config.PurchaseSuppression{ExecutionID: uuid.NewString(), Provider: "aws", Service: "rds", Region: "us-east-1",
+			ResourceType: "db.r5.large", SuppressedCount: 1, ExpiresAt: time.Now().Add(time.Hour)}
+		err = store.CreateSuppression(ctx, sup)
+		assert.True(t, errors.Is(err, config.ErrAZConfigMigrationNotReady), "CreateSuppression: %v", err)
+		err = store.WithTx(ctx, func(tx pgx.Tx) error { return store.CreateSuppressionTx(ctx, tx, sup) })
+		assert.True(t, errors.Is(err, config.ErrAZConfigMigrationNotReady), "CreateSuppressionTx: %v", err)
+		store.txs = 0 // the direct WithTx above is not the purchase path
+	}
+
+	for _, version := range []uint{155, 156} {
+		require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, version))
+		require.NoError(t, h.requireAZConfigSchema(ctx), "version %d", version)
+		_, err := store.ListActiveSuppressions(ctx)
+		require.NoError(t, err, "version %d", version)
+	}
+}

internal/api/handler_purchases.go

@@ -2833,6 +2833,9 @@ func (h *Handler) persistRetryExecution(ctx context.Context, failedExec *config.
 // The successor is a web-source pending row for the retrying user, so it
 // also goes through the submit dedupe (issue #760): see writeRetrySuccessorTx.
 func (h *Handler) persistRetrySuccessorTx(ctx context.Context, failedExec, newExecution *config.PurchaseExecution, suppressions []config.PurchaseSuppression) error {
+	if err := h.requireAZConfigSchema(ctx); err != nil {
+		return err
+	}
 	err := h.config.WithTx(ctx, func(tx pgx.Tx) error {
 		return h.writeRetrySuccessorTx(ctx, tx, failedExec, newExecution, suppressions)
 	})
@@ -3678,6 +3681,9 @@ func (h *Handler) persistExecutionAndSuppressions(
 	suppressions []config.PurchaseSuppression,
 	creatorID, idempotencyKey string,
 ) (dup *config.PurchaseExecution, err error) {
+	if err := h.requireAZConfigSchema(ctx); err != nil {
+		return nil, err
+	}
 	txErr := h.config.WithTx(ctx, func(tx pgx.Tx) error {
 		// Duplicate check inside the tx, after taking the creator's submit
 		// lock: a concurrent identical submit waits for this tx to commit,
@@ -3736,11 +3742,17 @@ func purchaseIdempotencyKey(creatorID string, recs []config.RecommendationRecord
 		if err != nil {
 			return "", err
 		}
-		tuples = append(tuples, fmt.Sprintf("%s|%s|%s|%s|%s|%s|%d|%d|%s|%s|%s|%s",
+		// Single-AZ and Multi-AZ RDS buys of the same class are different
+		// purchases and must not dedupe into one (issue #408).
+		azConfig, err := r.IdentityAZConfig()
+		if err != nil {
+			return "", err
+		}
+		tuples = append(tuples, fmt.Sprintf("%s|%s|%s|%s|%s|%s|%d|%d|%s|%s|%s|%s|%s",
 			strings.ToLower(strings.TrimSpace(r.Provider)),
 			r.Service, r.Region, r.ResourceType, r.Engine, acct,
 			r.Count, r.Term, strings.ToLower(strings.TrimSpace(r.Payment)), az,
-			ci.Platform, ci.Tenancy))
+			ci.Platform, ci.Tenancy, azConfig))
 	}
 	sort.Strings(tuples)
 	h := sha256.New()

internal/api/purchase_azconfig_test.go

@@ -0,0 +1,98 @@
+package api
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Real RDS Details shape as persisted by the scheduler (issue #408).
+func rdsDetails(azConfig string) json.RawMessage {
+	return json.RawMessage(`{"engine":"mysql","az_config":"` + azConfig + `","instance_class":"db.r5.large"}`)
+}
+
+func rdsAZConfigRec(azConfig string, upfront float64) config.RecommendationRecord {
+	return config.RecommendationRecord{
+		ID: "rec-rds-" + azConfig, Provider: "aws", Service: "rds", Region: "us-east-1", ResourceType: "db.r5.large",
+		Engine: "mysql", Term: 1, Payment: "all-upfront", Count: 1, UpfrontCost: upfront, Savings: 10,
+		Details: rdsDetails(azConfig),
+	}
+}
+
+// Stored Single-AZ and Multi-AZ rows of the same class must both load (no
+// "more than one row for identity key" refusal) and the request, which carries
+// az_config only in Details, is priced from its own variant's row.
+func TestPriceRecommendationsFromStore_PricesFromRequestedAZConfig(t *testing.T) {
+	for _, want := range []struct {
+		azConfig string
+		upfront  float64
+	}{
+		{"single-az", 100},
+		{"multi-az", 300},
+	} {
+		t.Run(want.azConfig, func(t *testing.T) {
+			store := new(MockConfigStore)
+			t.Cleanup(func() { store.AssertExpectations(t) })
+			expectStoredRecs(store, rdsAZConfigRec("single-az", 100), rdsAZConfigRec("multi-az", 300))
+			h := &Handler{config: store}
+			req := []config.RecommendationRecord{{
+				Provider: "aws", Service: "rds", Region: "us-east-1", ResourceType: "db.r5.large", Engine: "mysql",
+				Term: 1, Payment: "all-upfront", Count: 1, Details: rdsDetails(want.azConfig),
+			}}
+			require.NoError(t, h.priceRecommendationsFromStore(context.Background(), req))
+			assert.Equal(t, want.upfront, req[0].UpfrontCost)
+			got, err := req[0].DetailsAZConfig()
+			require.NoError(t, err)
+			assert.Equal(t, want.azConfig, got)
+		})
+	}
+}
+
+// A request that does not name its az_config cannot be priced off an arbitrary
+// RDS variant: it is refused with a 409.
+func TestPriceRecommendationsFromStore_RequestWithoutAZConfigRefusedForRDS(t *testing.T) {
+	store := new(MockConfigStore)
+	t.Cleanup(func() { store.AssertExpectations(t) })
+	expectStoredRecs(store, rdsAZConfigRec("multi-az", 300))
+	h := &Handler{config: store}
+	req := []config.RecommendationRecord{{
+		Provider: "aws", Service: "rds", Region: "us-east-1", ResourceType: "db.r5.large", Engine: "mysql",
+		Term: 1, Payment: "all-upfront", Count: 1,
+	}}
+	err := h.priceRecommendationsFromStore(context.Background(), req)
+	require.Error(t, err)
+	ce, isClient := IsClientError(err)
+	require.True(t, isClient)
+	assert.Equal(t, 409, ce.code)
+}
+
+func TestPurchaseIdempotencyKey_DiffersByAZConfig(t *testing.T) {
+	t.Parallel()
+	key := func(azConfig string) string {
+		return mustPurchaseIdempotencyKey("u", []config.RecommendationRecord{rdsAZConfigRec(azConfig, 1)}, 100)
+	}
+	assert.NotEqual(t, key("single-az"), key("multi-az"),
+		"a Multi-AZ submit right after a Single-AZ one must not dedupe into it")
+	assert.Equal(t, key("multi-az"), key("multi-az"))
+}
+
+func TestDescribeRec_IncludesAZConfig(t *testing.T) {
+	r := rdsAZConfigRec("multi-az", 1)
+	assert.Contains(t, describeRec(&r), `az_config="multi-az"`)
+	bare := config.RecommendationRecord{Provider: "aws", Service: "rds"}
+	assert.NotContains(t, describeRec(&bare), "az_config=")
+}
+
+func TestPurchaseIdempotencyKey_IgnoresAZConfigForAzureSQL(t *testing.T) {
+	t.Parallel()
+	key := func(azConfig string) string {
+		r := rdsAZConfigRec(azConfig, 1)
+		r.Provider, r.Service = "azure", "database"
+		return mustPurchaseIdempotencyKey("u", []config.RecommendationRecord{r}, 100)
+	}
+	assert.Equal(t, key(""), key("zoneRedundant"))
+}

internal/api/purchase_pricing.go

@@ -363,5 +363,8 @@ func describeRec(rec *config.RecommendationRecord) string {
 	if ci, err := rec.ComputeIdentity(); err == nil && !ci.IsZero() {
 		desc += fmt.Sprintf(" platform=%q tenancy=%q", ci.Platform, ci.Tenancy)
 	}
+	if azConfig, err := rec.DetailsAZConfig(); err == nil && azConfig != "" {
+		desc += fmt.Sprintf(" az_config=%q", azConfig)
+	}
 	return desc
 }

internal/config/errors.go

@@ -4,6 +4,11 @@ import "errors"
 
 var ErrAzurePricingNotReady = errors.New("azure pricing migration not ready")
 
+// ErrAZConfigMigrationNotReady means migrations 000154 (recommendations
+// az_config) and 000155 (purchase_suppressions az_config) are not both applied.
+// Apply them before running a release that reads or writes az_config.
+var ErrAZConfigMigrationNotReady = errors.New("az_config migrations 000154 and 000155 are not applied")
+
 // ErrNotFound is returned when a requested config-store row does not exist.
 var ErrNotFound = errors.New("not found")