diff --git a/internal/api/coverage_gaps_test.go b/internal/api/coverage_gaps_test.go index 62f415554..a81b1a50c 100644 --- a/internal/api/coverage_gaps_test.go +++ b/internal/api/coverage_gaps_test.go @@ -131,13 +131,7 @@ func TestRedactEmail(t *testing.T) { // --------------------------------------------------------------------------- func TestMergeServiceConfig_NewRecord(t *testing.T) { - ctx := context.Background() - mockStore := new(MockConfigStore) - - // Simulate "not found" error - mockStore.On("GetServiceConfig", ctx, "aws", "rds"). - Return(nil, errors.New("not found")) - + // No stored row: the request is the whole row. incoming := config.ServiceConfig{ Provider: "aws", Service: "rds", @@ -146,15 +140,12 @@ func TestMergeServiceConfig_NewRecord(t *testing.T) { Coverage: 80, } - result, err := mergeServiceConfig(ctx, mockStore, incoming, `{"enabled":true,"term":3,"coverage":80}`) + result, err := mergeServiceConfig(nil, incoming, `{"enabled":true,"term":3,"coverage":80}`) require.NoError(t, err) assert.Equal(t, incoming, result) } func TestMergeServiceConfig_ExistingRecord(t *testing.T) { - ctx := context.Background() - mockStore := new(MockConfigStore) - existing := &config.ServiceConfig{ Provider: "aws", Service: "rds", @@ -163,7 +154,6 @@ func TestMergeServiceConfig_ExistingRecord(t *testing.T) { Coverage: 50.0, IncludeEngines: []string{"mysql"}, } - mockStore.On("GetServiceConfig", ctx, "aws", "rds").Return(existing, nil) incoming := config.ServiceConfig{ Provider: "aws", @@ -173,7 +163,7 @@ func TestMergeServiceConfig_ExistingRecord(t *testing.T) { Coverage: 80.0, } - result, err := mergeServiceConfig(ctx, mockStore, incoming, `{"enabled":true,"term":3,"coverage":80}`) + result, err := mergeServiceConfig(existing, incoming, `{"enabled":true,"term":3,"coverage":80}`) require.NoError(t, err) // UI-editable fields are updated @@ -182,39 +172,19 @@ func TestMergeServiceConfig_ExistingRecord(t *testing.T) { assert.Equal(t, 80.0, result.Coverage) // Non-UI fields are preserved assert.Equal(t, []string{"mysql"}, result.IncludeEngines) + // The stored row the store handed over is left as it was. + assert.False(t, existing.Enabled) + assert.Equal(t, 1, existing.Term) + assert.Equal(t, 50.0, existing.Coverage) } -func TestMergeServiceConfig_DBError(t *testing.T) { - ctx := context.Background() - mockStore := new(MockConfigStore) - - mockStore.On("GetServiceConfig", ctx, "aws", "rds"). - Return(nil, errors.New("connection refused")) - - incoming := config.ServiceConfig{Provider: "aws", Service: "rds"} - - _, err := mergeServiceConfig(ctx, mockStore, incoming, `{"provider":"aws","service":"rds"}`) - assert.Error(t, err) - assert.Contains(t, err.Error(), "failed to read existing service config") -} +func TestMergeServiceConfig_MalformedBody(t *testing.T) { + existing := &config.ServiceConfig{Provider: "aws", Service: "rds"} -func TestMergeServiceConfig_NilExisting(t *testing.T) { - ctx := context.Background() - mockStore := new(MockConfigStore) - - // GetServiceConfig returns nil, nil (record does not exist but no error) - mockStore.On("GetServiceConfig", ctx, "aws", "rds").Return(nil, nil) - - incoming := config.ServiceConfig{ - Provider: "aws", - Service: "rds", - Enabled: true, - Term: 3, - } - - result, err := mergeServiceConfig(ctx, mockStore, incoming, `{"enabled":true,"term":3}`) - require.NoError(t, err) - assert.Equal(t, incoming, result) + _, err := mergeServiceConfig(existing, config.ServiceConfig{Provider: "aws", Service: "rds"}, `not json`) + var ce *clientError + require.ErrorAs(t, err, &ce) + assert.Equal(t, 400, ce.code) } // --------------------------------------------------------------------------- diff --git a/internal/api/handler_config.go b/internal/api/handler_config.go index 182e2425a..c519c048a 100644 --- a/internal/api/handler_config.go +++ b/internal/api/handler_config.go @@ -176,14 +176,70 @@ func (h *Handler) checkCommitmentOptionCombo(ctx context.Context, cfg config.Ser return nil } if err != nil { - logging.Errorf("commitment-option validation error (refusing save): %v", err) - return NewClientError(503, "could not validate the commitment term and payment option; try again") + return commitmentComboUnverifiable(err) } if !ok { - return NewClientError(400, fmt.Sprintf( - "%s does not support %dyr %s commitments", - cfg.Service, cfg.Term, cfg.Payment, - )) + return commitmentComboUnsupported(cfg) + } + return nil +} + +// commitmentComboUnverifiable is the 503 a save gets when the commitment +// options could not be read, so the combo could not be checked. +func commitmentComboUnverifiable(err error) error { + logging.Errorf("commitment-option validation error (refusing save): %v", err) + return NewClientError(503, "could not validate the commitment term and payment option; try again") +} + +// commitmentComboUnsupported is the 400 a save gets for a (term, payment) +// combination the probe data says the service does not sell. +func commitmentComboUnsupported(cfg config.ServiceConfig) error { + return NewClientError(400, fmt.Sprintf( + "%s does not support %dyr %s commitments", + cfg.Service, cfg.Term, cfg.Payment, + )) +} + +// commitmentOptsSnapshot is the commitment-option data a service-config PUT +// checks its merged row against. It is fetched before the save opens its +// transaction (issue #1488): Get reads the store through pooled connections +// and can probe AWS on a cold cache, and doing that while the transaction +// holds a pooled connection deadlocks the pool once as many saves run at +// once as the pool has connections (the Settings page saves its rows in +// parallel). check is pure, so it is safe inside the transaction. +type commitmentOptsSnapshot struct { + opts commitmentopts.Options + err error + // loaded is false when there is nothing to check against: no service is + // wired, the save is not AWS, or no probe data exists (ErrNoData). + loaded bool +} + +// prefetchCommitmentOptions fetches the snapshot for a save of provider. A +// read error is kept rather than returned, so it fails only a save whose +// merged row actually needs the check, as checkCommitmentOptionCombo does. +func (h *Handler) prefetchCommitmentOptions(ctx context.Context, provider string) commitmentOptsSnapshot { + if h.commitmentOpts == nil || provider != "aws" { + return commitmentOptsSnapshot{} + } + opts, err := h.commitmentOpts.Get(ctx) + if errors.Is(err, commitmentopts.ErrNoData) { + return commitmentOptsSnapshot{} + } + return commitmentOptsSnapshot{opts: opts, err: err, loaded: true} +} + +// check applies checkCommitmentOptionCombo's rules to cfg against the +// snapshot, without touching the database or the network. +func (s commitmentOptsSnapshot) check(cfg config.ServiceConfig) error { + if !s.loaded || cfg.Provider != "aws" || cfg.Term <= 0 || cfg.Payment == "" { + return nil + } + if s.err != nil { + return commitmentComboUnverifiable(s.err) + } + if !commitmentopts.ValidateCombo(s.opts, cfg.Provider, cfg.Service, cfg.Term, cfg.Payment) { + return commitmentComboUnsupported(cfg) } return nil } @@ -203,36 +259,29 @@ var serviceConfigOverlayKeys = []string{ "min_count", } -// mergeServiceConfig loads any existing service config and overlays the -// UI-editable fields from cfg onto it. Every field (Enabled, Term, Payment, -// Coverage and the include/exclude engines/regions/types and min_count -// filters) is overlaid only when the request body actually carried the -// corresponding key (per presentKeys), so a partial PUT cannot silently -// revert or wipe a field it did not mean to touch. body is the raw request -// JSON used to detect which keys are present. +// mergeServiceConfig overlays the UI-editable fields from cfg onto the stored +// row existing. Every field (Enabled, Term, Payment, Coverage and the +// include/exclude engines/regions/types and min_count filters) is overlaid +// only when the request body actually carried the corresponding key (per +// presentKeys), so a partial PUT cannot silently revert or wipe a field it +// did not mean to touch. body is the raw request JSON used to detect which +// keys are present. // -// A "not found" error means no existing record — cfg is returned unchanged. -// Any other DB error is returned to prevent a partial write from clobbering -// previously configured filter fields. -func mergeServiceConfig(ctx context.Context, store config.StoreInterface, cfg config.ServiceConfig, body string) (config.ServiceConfig, error) { - existing, err := store.GetServiceConfig(ctx, cfg.Provider, cfg.Service) - if err != nil { - if strings.Contains(err.Error(), "not found") { - return cfg, nil // new record — no existing fields to preserve - } - return cfg, fmt.Errorf("failed to read existing service config before update: %w", err) - } +// A nil existing means no stored row: cfg is returned unchanged. existing is +// not modified. +func mergeServiceConfig(existing *config.ServiceConfig, cfg config.ServiceConfig, body string) (config.ServiceConfig, error) { if existing == nil { - return cfg, nil + return cfg, nil // new record — no existing fields to preserve } present, perr := presentKeys(body, serviceConfigOverlayKeys) if perr != nil { return cfg, perr } - overlayPresentScalarFields(existing, &cfg, present) - overlayPresentFilterFields(existing, &cfg, present) - return *existing, nil + merged := *existing + overlayPresentScalarFields(&merged, &cfg, present) + overlayPresentFilterFields(&merged, &cfg, present) + return merged, nil } // presentKeys returns the subset of keys that appear as top-level fields in the @@ -360,28 +409,44 @@ func (h *Handler) updateServiceConfig(ctx context.Context, req *events.LambdaFun cfg.Service = parts[1] } - // Merge: overlay each field (the scalars and the recommendation filters) - // only when the body carried its key, so a partial PUT never reverts or - // zeroes a field (ramp_schedule, a filter set out-of-band) it didn't mean - // to touch. Validation below runs on the merged row. - merged, mergeErr := mergeServiceConfig(ctx, h.config, cfg, req.Body) - if mergeErr != nil { - return nil, mergeErr - } - cfg = merged + // Fetch the commitment options before the transaction opens: the merge + // callback below runs while the transaction holds a pooled connection, + // so nothing in it may take another one (see commitmentOptsSnapshot). + combos := h.prefetchCommitmentOptions(ctx, cfg.Provider) - // Validate the configuration - if err := cfg.Validate(); err != nil { - return nil, NewClientError(400, fmt.Sprintf("validation error: %s", err)) + // Serialized read-merge-write (issue #1488): the store reads the stored + // row and writes the merged result in one locked transaction, so two + // admins saving different fields of this row at once cannot both merge + // onto the same stale base and lose one change. ClientErrors from the + // merge or validation propagate unchanged. + _, err := h.config.UpdateServiceConfigAtomic(ctx, cfg.Provider, cfg.Service, func(existing *config.ServiceConfig) (*config.ServiceConfig, error) { + return mergeAndValidateServiceConfig(existing, cfg, req.Body, combos) + }) + if err != nil { + return nil, err } - if err := h.checkCommitmentOptionCombo(ctx, cfg); err != nil { + return &StatusResponse{Status: "updated"}, nil +} + +// mergeAndValidateServiceConfig is the read-merge step of a service-config +// PUT, run inside UpdateServiceConfigAtomic's transaction. It overlays each +// field (the scalars and the recommendation filters) only when the body +// carried its key, so a partial PUT never reverts or zeroes a field +// (ramp_schedule, a filter set out-of-band) it didn't mean to touch, then +// validates the merged row, not the partial body, including its commitment +// combo against the prefetched combos. It touches neither the database nor +// the network: the transaction already holds a pooled connection. +func mergeAndValidateServiceConfig(existing *config.ServiceConfig, cfg config.ServiceConfig, body string, combos commitmentOptsSnapshot) (*config.ServiceConfig, error) { + merged, err := mergeServiceConfig(existing, cfg, body) + if err != nil { return nil, err } - - if err := h.config.SaveServiceConfig(ctx, &cfg); err != nil { + if err := merged.Validate(); err != nil { + return nil, NewClientError(400, fmt.Sprintf("validation error: %s", err)) + } + if err := combos.check(merged); err != nil { return nil, err } - - return &StatusResponse{Status: "updated"}, nil + return &merged, nil } diff --git a/internal/api/handler_config_service_race_integration_test.go b/internal/api/handler_config_service_race_integration_test.go new file mode 100644 index 000000000..a2b20aa5c --- /dev/null +++ b/internal/api/handler_config_service_race_integration_test.go @@ -0,0 +1,205 @@ +//go:build integration + +package api + +// Real-Postgres race tests for the service-config PUT (issue #1488). The PUT +// reads the stored row, overlays only the keys the body carries and writes the +// whole row back. Unless that read-merge-write is one locked transaction, two +// admins saving different fields of the same row at the same moment both read +// the same base and the later write silently drops the earlier change. A mock +// cannot hold a row lock, so these run against a real database. + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/commitmentopts" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/aws/aws-lambda-go/events" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// serviceConfigRaceRounds is how many concurrent PUT pairs each test fires. +// Before the fix one update was lost in nearly every round, so a handful +// would do; more rounds make a regression all but certain to show. +const serviceConfigRaceRounds = 40 + +type serviceConfigRaceFixture struct { + pg *testhelpers.PostgresContainer + store *config.PostgresStore + handler *Handler +} + +func newServiceConfigRaceFixture(ctx context.Context, t *testing.T) *serviceConfigRaceFixture { + t.Helper() + pg := testhelpers.RequireMigratedDatabase(ctx, t, getMigrationsPath()) + store := config.NewPostgresStore(pg.DB) + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", ctx, "admin-token").Return(&Session{ + UserID: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", Email: "admin@example.com", + }, nil) + mockAuth.grantAdmin() + return &serviceConfigRaceFixture{pg: pg, store: store, handler: &Handler{config: store, auth: mockAuth}} +} + +// putConcurrently fires every body as a PUT to aws/rds at the same moment and +// fails the test if any of them errors. +func (f *serviceConfigRaceFixture) putConcurrently(ctx context.Context, t *testing.T, bodies ...string) { + t.Helper() + start := make(chan struct{}) + errs := make([]error, len(bodies)) + var wg sync.WaitGroup + for i, body := range bodies { + wg.Add(1) + go func(i int, body string) { + defer wg.Done() + <-start + _, errs[i] = f.handler.updateServiceConfig(ctx, &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"Authorization": "Bearer admin-token"}, + Body: body, + }, "aws/rds") + }(i, body) + } + close(start) + wg.Wait() + for i, err := range errs { + require.NoError(t, err, "PUT %d (%s)", i, bodies[i]) + } +} + +// TestUpdateServiceConfig_ConcurrentPartialPutsKeepBothKeys is the issue +// #1488 scenario: two PUTs of different keys on an existing row, at once. +// Both changes must persist, along with the fields neither PUT touched. +func TestUpdateServiceConfig_ConcurrentPartialPutsKeepBothKeys(t *testing.T) { + ctx := context.Background() + f := newServiceConfigRaceFixture(ctx, t) + + lost := 0 + for round := 1; round <= serviceConfigRaceRounds; round++ { + require.NoError(t, f.store.SaveServiceConfig(ctx, &config.ServiceConfig{ + Provider: "aws", Service: "rds", Enabled: true, Term: 3, Payment: "all-upfront", + Coverage: 50, RampSchedule: config.RampImmediate, IncludeEngines: []string{"mysql"}, + })) + coverage, minCount := float64(10+round), round + + f.putConcurrently(ctx, t, + fmt.Sprintf(`{"coverage":%v}`, coverage), + fmt.Sprintf(`{"min_count":%d}`, minCount)) + + got, err := f.store.GetServiceConfig(ctx, "aws", "rds") + require.NoError(t, err) + if got.Coverage != coverage || got.MinCount != minCount { + lost++ + t.Logf("round %d: coverage=%v (want %v) min_count=%d (want %d)", round, got.Coverage, coverage, got.MinCount, minCount) + } + require.Equal(t, []any{true, 3, "all-upfront", []string{"mysql"}}, + []any{got.Enabled, got.Term, got.Payment, got.IncludeEngines}, "round %d: untouched fields must survive", round) + } + require.Zero(t, lost, "%d of %d rounds lost one of two concurrent partial PUTs", lost, serviceConfigRaceRounds) +} + +// TestUpdateServiceConfig_ConcurrentFirstSavesKeepBothKeys covers the same race +// when no row exists yet: a row lock cannot serialize two saves that both find +// nothing to lock, so each would insert its own full row and the second would +// overwrite the first. Both changes must persist. +func TestUpdateServiceConfig_ConcurrentFirstSavesKeepBothKeys(t *testing.T) { + ctx := context.Background() + f := newServiceConfigRaceFixture(ctx, t) + + lost := 0 + for round := 1; round <= serviceConfigRaceRounds; round++ { + _, err := f.pg.DB.Pool().Exec(ctx, `DELETE FROM service_configs WHERE provider = 'aws' AND service = 'rds'`) + require.NoError(t, err) + coverage, minCount := float64(10+round), round + + f.putConcurrently(ctx, t, + fmt.Sprintf(`{"term":1,"payment":"no-upfront","coverage":%v}`, coverage), + fmt.Sprintf(`{"term":1,"payment":"no-upfront","min_count":%d}`, minCount)) + + got, err := f.store.GetServiceConfig(ctx, "aws", "rds") + require.NoError(t, err) + if got.Coverage != coverage || got.MinCount != minCount { + lost++ + t.Logf("round %d: coverage=%v (want %v) min_count=%d (want %d)", round, got.Coverage, coverage, got.MinCount, minCount) + } + } + require.Zero(t, lost, "%d of %d rounds lost one of two concurrent first saves", lost, serviceConfigRaceRounds) +} + +// settingsAWSServices are the AWS rows the Settings page saves, one PUT each, +// in parallel (frontend/src/settings.ts, Promise.allSettled). +var settingsAWSServices = []string{ + "ec2", "rds", "elasticache", "opensearch", "redshift", + "savings-plans-compute", "savings-plans-ec2instance", "savings-plans-sagemaker", "savings-plans-database", +} + +// TestUpdateServiceConfig_ParallelSettingsSaveDoesNotExhaustPool is the pool +// deadlock from the #1509 review. Each PUT holds one pooled connection for its +// locked transaction. If the commitment-option check inside it reads the +// commitmentopts store through the same pool, then as many concurrent saves as +// the pool has connections each hold one and wait for a second that never +// frees: every save hangs until its deadline and the instance starves. The +// pool here is the Fargate default (db_max_connections = 3), the probe cache is +// warm, and the handler is wired as internal/server/app.go wires it. +func TestUpdateServiceConfig_ParallelSettingsSaveDoesNotExhaustPool(t *testing.T) { + ctx := context.Background() + // The limit under test is this client pool's size, so the shared server + // is fine. + pg := testhelpers.RequireMigratedDatabase(ctx, t, getMigrationsPath()) + poolCfg := *pg.Config + poolCfg.MaxConnections, poolCfg.MinConnections = 3, 0 + db, err := database.NewConnection(ctx, &poolCfg, nil) + require.NoError(t, err) + t.Cleanup(db.Close) + + optsStore := commitmentopts.NewPostgresStore(db) + combos := make([]commitmentopts.Combo, 0, len(settingsAWSServices)) + for _, svc := range settingsAWSServices { + combos = append(combos, commitmentopts.Combo{Provider: "aws", Service: svc, TermYears: 1, Payment: "no-upfront"}) + } + require.NoError(t, optsStore.Save(ctx, combos, "123456789012")) + + store := config.NewPostgresStore(db) + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", mock.Anything, "admin-token").Return(&Session{ + UserID: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", Email: "admin@example.com", + }, nil) + mockAuth.grantAdmin() + h := &Handler{config: store, auth: mockAuth, commitmentOpts: commitmentopts.New(optsStore, store, nil, nil)} + + // Before the fix every save fails here at the deadline; after it the + // whole batch takes milliseconds. + reqCtx, cancel := context.WithTimeout(ctx, 15*time.Second) + defer cancel() + start := make(chan struct{}) + errs := make([]error, len(settingsAWSServices)) + var wg sync.WaitGroup + for i, svc := range settingsAWSServices { + wg.Add(1) + go func(i int, svc string) { + defer wg.Done() + <-start + _, errs[i] = h.updateServiceConfig(reqCtx, &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"Authorization": "Bearer admin-token"}, + Body: `{"enabled":true,"term":1,"payment":"no-upfront","coverage":50}`, + }, "aws/"+svc) + }(i, svc) + } + began := time.Now() + close(start) + wg.Wait() + t.Logf("%d parallel saves on a pool of %d took %v", len(settingsAWSServices), poolCfg.MaxConnections, time.Since(began)) + + for i, svc := range settingsAWSServices { + require.NoError(t, errs[i], "PUT aws/%s", svc) + got, err := store.GetServiceConfig(ctx, "aws", svc) + require.NoError(t, err) + require.Equal(t, []any{true, 1, "no-upfront", 50.0}, []any{got.Enabled, got.Term, got.Payment, got.Coverage}, "aws/%s", svc) + } +} diff --git a/internal/api/handler_config_test.go b/internal/api/handler_config_test.go index c89d073de..8c02c371a 100644 --- a/internal/api/handler_config_test.go +++ b/internal/api/handler_config_test.go @@ -7,6 +7,7 @@ import ( "sync" "testing" + "github.com/LeanerCloud/cloud-commitments-platform/internal/commitmentopts" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" "github.com/aws/aws-lambda-go/events" "github.com/stretchr/testify/assert" @@ -645,11 +646,8 @@ func TestHandler_updateServiceConfig(t *testing.T) { // overlaid. A partial PUT that omits a filter must preserve the existing // value; a PUT that includes it (even empty) must apply it. func TestMergeServiceConfig_PresenceAwareFilterOverlay(t *testing.T) { - ctx := context.Background() - // Each subtest gets a fresh existing record: mergeServiceConfig overlays - // onto the pointer returned by GetServiceConfig (the production postgres - // store returns a fresh struct per call), so a shared fixture would leak - // mutations across subtests. + // Each subtest gets a fresh existing record so no subtest can see + // another's data. newExisting := func() *config.ServiceConfig { return &config.ServiceConfig{ Provider: "aws", Service: "rds", Enabled: true, Term: 3, @@ -661,10 +659,6 @@ func TestMergeServiceConfig_PresenceAwareFilterOverlay(t *testing.T) { } t.Run("body includes filter fields -> overlaid", func(t *testing.T) { - store := new(MockConfigStore) - store.On("GetServiceConfig", ctx, "aws", "rds").Return(newExisting(), nil) - t.Cleanup(func() { store.AssertExpectations(t) }) - req := config.ServiceConfig{ Provider: "aws", Service: "rds", Enabled: false, Term: 1, Payment: "no-upfront", Coverage: 90, @@ -673,7 +667,7 @@ func TestMergeServiceConfig_PresenceAwareFilterOverlay(t *testing.T) { } body := `{"enabled":false,"term":1,"payment":"no-upfront","coverage":90,"include_engines":["postgres"],"min_count":7}` - merged, err := mergeServiceConfig(ctx, store, req, body) + merged, err := mergeServiceConfig(newExisting(), req, body) require.NoError(t, err) assert.False(t, merged.Enabled) assert.Equal(t, 1, merged.Term) @@ -685,17 +679,13 @@ func TestMergeServiceConfig_PresenceAwareFilterOverlay(t *testing.T) { }) t.Run("body omits filter fields -> preserved", func(t *testing.T) { - store := new(MockConfigStore) - store.On("GetServiceConfig", ctx, "aws", "rds").Return(newExisting(), nil) - t.Cleanup(func() { store.AssertExpectations(t) }) - req := config.ServiceConfig{ Provider: "aws", Service: "rds", Enabled: true, Term: 3, Payment: "all-upfront", Coverage: 80, } body := `{"enabled":true,"term":3,"payment":"all-upfront","coverage":80}` - merged, err := mergeServiceConfig(ctx, store, req, body) + merged, err := mergeServiceConfig(newExisting(), req, body) require.NoError(t, err) assert.Equal(t, 80.0, merged.Coverage) assert.Equal(t, []string{"mysql"}, merged.IncludeEngines, "omitted filter must be preserved") @@ -704,10 +694,6 @@ func TestMergeServiceConfig_PresenceAwareFilterOverlay(t *testing.T) { }) t.Run("body includes empty filter -> cleared", func(t *testing.T) { - store := new(MockConfigStore) - store.On("GetServiceConfig", ctx, "aws", "rds").Return(newExisting(), nil) - t.Cleanup(func() { store.AssertExpectations(t) }) - req := config.ServiceConfig{ Provider: "aws", Service: "rds", Enabled: true, Term: 3, Payment: "all-upfront", Coverage: 80, @@ -715,7 +701,7 @@ func TestMergeServiceConfig_PresenceAwareFilterOverlay(t *testing.T) { } body := `{"enabled":true,"term":3,"payment":"all-upfront","coverage":80,"include_engines":[]}` - merged, err := mergeServiceConfig(ctx, store, req, body) + merged, err := mergeServiceConfig(newExisting(), req, body) require.NoError(t, err) assert.Empty(t, merged.IncludeEngines, "explicit empty list clears the filter") }) @@ -726,7 +712,6 @@ func TestMergeServiceConfig_PresenceAwareFilterOverlay(t *testing.T) { // body carries the key, so a stale tab that PUTs just the field the user // edited cannot revert the others (#539). A present zero value still applies. func TestMergeServiceConfig_PresenceAwareScalarOverlay(t *testing.T) { - ctx := context.Background() newExisting := func() *config.ServiceConfig { return &config.ServiceConfig{ Provider: "aws", Service: "rds", Enabled: true, Term: 3, @@ -735,11 +720,8 @@ func TestMergeServiceConfig_PresenceAwareScalarOverlay(t *testing.T) { } run := func(t *testing.T, req config.ServiceConfig, body string) config.ServiceConfig { t.Helper() - store := new(MockConfigStore) - store.On("GetServiceConfig", ctx, "aws", "rds").Return(newExisting(), nil) - t.Cleanup(func() { store.AssertExpectations(t) }) req.Provider, req.Service = "aws", "rds" - merged, err := mergeServiceConfig(ctx, store, req, body) + merged, err := mergeServiceConfig(newExisting(), req, body) require.NoError(t, err) return merged } @@ -828,6 +810,153 @@ func TestHandler_updateServiceConfig_PartialPut(t *testing.T) { }) } +// TestHandler_updateServiceConfig_MergesInsideAtomicUpdate pins issue #1488: +// the PUT must read and merge the stored row inside the store's locked +// UpdateServiceConfigAtomic, never through a separate GetServiceConfig + +// SaveServiceConfig pair that two concurrent PUTs can interleave. +func TestHandler_updateServiceConfig_MergesInsideAtomicUpdate(t *testing.T) { + ctx := context.Background() + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", ctx, "admin-token").Return(&Session{ + UserID: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", Email: "admin@example.com", + }, nil) + mockAuth.grantAdmin() + store := new(MockConfigStore) + stored := &config.ServiceConfig{ + Provider: "aws", Service: "rds", Enabled: true, Term: 3, + Payment: "all-upfront", Coverage: 55, MinCount: 4, + } + var written *config.ServiceConfig + store.On("UpdateServiceConfigAtomic", ctx, "aws", "rds", mock.Anything). + Run(func(args mock.Arguments) { + apply := args.Get(3).(func(*config.ServiceConfig) (*config.ServiceConfig, error)) + var err error + written, err = apply(stored) + require.NoError(t, err) + }).Return(stored, nil) + h := &Handler{config: store, auth: mockAuth} + + _, err := h.updateServiceConfig(ctx, &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"Authorization": "Bearer admin-token"}, + Body: `{"coverage":80}`, + }, "aws/rds") + require.NoError(t, err) + require.NotNil(t, written) + assert.Equal(t, 80.0, written.Coverage) + assert.Equal(t, []any{true, 3, "all-upfront", 4}, []any{written.Enabled, written.Term, written.Payment, written.MinCount}) + store.AssertNotCalled(t, "GetServiceConfig", mock.Anything, mock.Anything, mock.Anything) + store.AssertNotCalled(t, "SaveServiceConfig", mock.Anything, mock.Anything) +} + +// TestHandler_updateServiceConfig_CommitmentOptsOutsideTransaction pins the +// pool-deadlock fix for issue #1488: the commitment options are fetched before +// UpdateServiceConfigAtomic opens its transaction, never from inside the merge +// callback (which would need a second pooled connection while the transaction +// holds one), and the callback checks the merged row against that snapshot. +func TestHandler_updateServiceConfig_CommitmentOptsOutsideTransaction(t *testing.T) { + ctx := context.Background() + rdsOneYear := commitmentopts.Options{"aws": {"rds": { + {Provider: "aws", Service: "rds", TermYears: 1, Payment: "all-upfront"}, + }}} + run := func(t *testing.T, stored *config.ServiceConfig, body string, getErr error) error { + t.Helper() + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", ctx, "admin-token").Return(&Session{ + UserID: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", Email: "admin@example.com", + }, nil) + mockAuth.grantAdmin() + // The mock's UpdateServiceConfigAtomic fallback reads the row through + // GetServiceConfig, then runs the merge callback, then saves: that + // read is the point where the real store has opened its transaction. + txOpened, gets := false, 0 + store := new(MockConfigStore) + store.On("GetServiceConfig", ctx, "aws", "rds").Return(stored, nil). + Run(func(mock.Arguments) { txOpened = true }) + store.On("SaveServiceConfig", ctx, mock.AnythingOfType("*config.ServiceConfig")).Return(nil) + h := &Handler{config: store, auth: mockAuth, commitmentOpts: &stubCommitmentOpts{ + getFn: func(context.Context) (commitmentopts.Options, error) { + assert.False(t, txOpened, "commitment options read inside the transaction") + gets++ + return rdsOneYear, getErr + }, + validateFn: func(context.Context, string, string, int, string) (bool, error) { + t.Error("Validate reads the store; the PUT must check the prefetched snapshot") + return true, nil + }, + }} + _, err := h.updateServiceConfig(ctx, &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"Authorization": "Bearer admin-token"}, Body: body, + }, "aws/rds") + assert.Equal(t, 1, gets, "options fetched exactly once, before the transaction") + return err + } + storedThreeYear := &config.ServiceConfig{Provider: "aws", Service: "rds", Enabled: true, Term: 3, Payment: "no-upfront", Coverage: 50} + + t.Run("merged row checked against the snapshot", func(t *testing.T) { + // The body alone carries no term or payment; the stored 3yr no-upfront + // does, and the snapshot does not sell it. + err := run(t, storedThreeYear, `{"coverage":80}`, nil) + ce, ok := IsClientError(err) + require.True(t, ok, "got %v", err) + assert.Equal(t, 400, ce.code) + assert.Contains(t, ce.message, "3yr no-upfront") + }) + t.Run("supported combo saves", func(t *testing.T) { + require.NoError(t, run(t, nil, `{"enabled":true,"term":1,"payment":"all-upfront","coverage":80}`, nil)) + }) + t.Run("read error refuses a save that needs the check", func(t *testing.T) { + err := run(t, storedThreeYear, `{"coverage":80}`, errors.New("connection refused")) + ce, ok := IsClientError(err) + require.True(t, ok, "got %v", err) + assert.Equal(t, 503, ce.code) + }) + t.Run("read error does not block a save without a combo", func(t *testing.T) { + require.NoError(t, run(t, nil, `{"enabled":false,"coverage":80}`, errors.New("connection refused"))) + }) +} + +// TestHandler_updateServiceConfig_StoredRowRead covers the two outcomes of the +// in-transaction read: no stored row saves the body as the whole row, and any +// other read error aborts the save. +func TestHandler_updateServiceConfig_StoredRowRead(t *testing.T) { + ctx := context.Background() + newHandler := func(readErr error) (*Handler, *MockConfigStore) { + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", ctx, "admin-token").Return(&Session{ + UserID: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", Email: "admin@example.com", + }, nil) + mockAuth.grantAdmin() + store := new(MockConfigStore) + store.On("GetServiceConfig", ctx, "aws", "rds").Return(nil, readErr) + return &Handler{config: store, auth: mockAuth}, store + } + put := func(h *Handler) error { + _, err := h.updateServiceConfig(ctx, &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"Authorization": "Bearer admin-token"}, + Body: `{"enabled":true,"term":1,"payment":"no-upfront","coverage":80}`, + }, "aws/rds") + return err + } + + t.Run("no stored row -> body saved as the row", func(t *testing.T) { + h, store := newHandler(fmt.Errorf("service config not found for aws:rds: %w", config.ErrNotFound)) + var saved *config.ServiceConfig + store.On("SaveServiceConfig", ctx, mock.AnythingOfType("*config.ServiceConfig")). + Run(func(args mock.Arguments) { saved = args.Get(1).(*config.ServiceConfig) }).Return(nil) + require.NoError(t, put(h)) + require.NotNil(t, saved) + assert.Equal(t, config.ServiceConfig{ + Provider: "aws", Service: "rds", Enabled: true, Term: 1, Payment: "no-upfront", Coverage: 80, + }, *saved) + }) + + t.Run("read error -> no save", func(t *testing.T) { + h, store := newHandler(errors.New("connection refused")) + require.ErrorContains(t, put(h), "connection refused") + store.AssertNotCalled(t, "SaveServiceConfig", mock.Anything, mock.Anything) + }) +} + func TestHandler_updateServiceConfig_InvalidBody(t *testing.T) { ctx := context.Background() mockAuth := new(MockAuthService) @@ -869,17 +998,19 @@ func TestHandler_updateServiceConfig_CommitmentOptsReject(t *testing.T) { mockStore.On("GetServiceConfig", ctx, "aws", "rds").Return(nil, nil) // Probe data says RDS 3yr no-upfront doesn't exist. Save must 400. - // SaveServiceConfig is NOT set up — asserting it's never called. + // SaveServiceConfig is NOT set up — asserting it's never called. The PUT + // prefetches the options with Get before its transaction and checks the + // merged row against them inside it, so validateFn stays nil: calling the + // pool-touching Validate there would panic this test (issue #1488). handler := &Handler{ config: mockStore, auth: mockAuth, commitmentOpts: &stubCommitmentOpts{ - validateFn: func(_ context.Context, provider, service string, term int, payment string) (bool, error) { - assert.Equal(t, "aws", provider) - assert.Equal(t, "rds", service) - assert.Equal(t, 3, term) - assert.Equal(t, "no-upfront", payment) - return false, nil + getFn: func(context.Context) (commitmentopts.Options, error) { + return commitmentopts.Options{"aws": {"rds": { + {Provider: "aws", Service: "rds", TermYears: 1, Payment: "all-upfront"}, + {Provider: "aws", Service: "rds", TermYears: 3, Payment: "all-upfront"}, + }}}, nil }, }, } @@ -918,8 +1049,10 @@ func TestHandler_updateServiceConfig_CommitmentOptsAccept(t *testing.T) { config: mockStore, auth: mockAuth, commitmentOpts: &stubCommitmentOpts{ - validateFn: func(context.Context, string, string, int, string) (bool, error) { - return true, nil + getFn: func(context.Context) (commitmentopts.Options, error) { + return commitmentopts.Options{"aws": {"rds": { + {Provider: "aws", Service: "rds", TermYears: 1, Payment: "all-upfront"}, + }}}, nil }, }, } @@ -952,8 +1085,8 @@ func TestHandler_updateServiceConfig_CommitmentOptsErrorRefusesSave(t *testing.T config: mockStore, auth: mockAuth, commitmentOpts: &stubCommitmentOpts{ - validateFn: func(context.Context, string, string, int, string) (bool, error) { - return false, errors.New("read commitmentopts store: connection refused") + getFn: func(context.Context) (commitmentopts.Options, error) { + return nil, errors.New("read commitmentopts store: connection refused") }, }, } diff --git a/internal/commitmentopts/service.go b/internal/commitmentopts/service.go index 60bbf8005..7f9dd8a1e 100644 --- a/internal/commitmentopts/service.go +++ b/internal/commitmentopts/service.go @@ -154,6 +154,11 @@ func (s *Service) findAWSAccount(ctx context.Context) (*config.CloudAccount, err // successfully probed) Validate returns true so saves aren't blocked when // we can't verify. The frontend's hardcoded rules are the user-facing // gate; this check is belt-and-braces. +// +// Validate reads the store through its own pool connections and can probe +// AWS on a cold cache, so it must not run while the caller holds a pooled +// transaction. Such callers fetch the Options with Get first and check them +// with ValidateCombo inside the transaction. func (s *Service) Validate(ctx context.Context, provider, service string, term int, payment string) (bool, error) { opts, err := s.Get(ctx) if err != nil { @@ -162,12 +167,21 @@ func (s *Service) Validate(ctx context.Context, provider, service string, term i } return false, err } + return ValidateCombo(opts, provider, service, term, payment), nil +} +// ValidateCombo reports whether (provider, service, term, payment) is a legal +// combination according to opts, a snapshot returned by Get. It is pure: no +// database or network access, so it is safe to call inside a transaction. +// +// It is permissive where opts has nothing to say: a provider with no combos, +// or a service the probe returned no combos for, is treated as valid. +func ValidateCombo(opts Options, provider, service string, term int, payment string) bool { byService, ok := opts[provider] if !ok { // No combos for this provider at all — nothing to validate // against. Permissive fallback keeps parity with ErrNoData. - return true, nil + return true } combos, ok := byService[service] if !ok { @@ -181,14 +195,14 @@ func (s *Service) Validate(ctx context.Context, provider, service string, term i // rules are the primary user-facing gate. logging.Warnf("commitmentopts: provider %q known in probe data but service %q absent; treating as valid (term=%d payment=%q)", provider, service, term, payment) - return true, nil + return true } for _, c := range combos { if c.TermYears == term && c.Payment == payment { - return true, nil + return true } } - return false, nil + return false } // buildOptions groups a flat Combo slice into the nested Options map the diff --git a/internal/commitmentopts/service_test.go b/internal/commitmentopts/service_test.go index 78e281f52..771f8beb4 100644 --- a/internal/commitmentopts/service_test.go +++ b/internal/commitmentopts/service_test.go @@ -227,6 +227,22 @@ func TestService_Validate_HitAndMiss(t *testing.T) { assert.False(t, ok) } +// ValidateCombo is the store-free check callers run inside a transaction +// against a snapshot fetched earlier with Get (issue #1488). It must give the +// same answers as Validate. +func TestValidateCombo_Snapshot(t *testing.T) { + opts := Options{"aws": {"rds": { + {Provider: "aws", Service: "rds", TermYears: 1, Payment: "all-upfront"}, + {Provider: "aws", Service: "rds", TermYears: 3, Payment: "no-upfront"}, + }}} + assert.True(t, ValidateCombo(opts, "aws", "rds", 1, "all-upfront")) + assert.True(t, ValidateCombo(opts, "aws", "rds", 3, "no-upfront")) + assert.False(t, ValidateCombo(opts, "aws", "rds", 3, "partial-upfront")) + assert.True(t, ValidateCombo(opts, "aws", "ec2", 3, "partial-upfront"), "service absent from the snapshot is permissive") + assert.True(t, ValidateCombo(opts, "azure", "vm", 1, "all-upfront"), "provider absent from the snapshot is permissive") + assert.True(t, ValidateCombo(nil, "aws", "rds", 3, "partial-upfront"), "nil snapshot is permissive") +} + func TestService_Validate_UnknownProviderPermissive(t *testing.T) { cached := Options{"aws": {"rds": {{Provider: "aws", Service: "rds", TermYears: 1, Payment: "all-upfront"}}}} store := &fakeStore{opts: cached, has: true} diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 20b8abb54..497923146 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -26,6 +26,12 @@ type StoreInterface interface { // Service configuration GetServiceConfig(ctx context.Context, provider, service string) (*ServiceConfig, error) SaveServiceConfig(ctx context.Context, config *ServiceConfig) error + // UpdateServiceConfigAtomic serializes a read-merge-write of one service + // config row under a locked transaction so concurrent partial PUTs of the + // same row cannot lose each other's updates. apply receives the stored row + // (nil when none exists) and returns the full row to write; its error + // aborts the write and is propagated. + UpdateServiceConfigAtomic(ctx context.Context, provider, service string, apply func(existing *ServiceConfig) (*ServiceConfig, error)) (*ServiceConfig, error) ListServiceConfigs(ctx context.Context) ([]ServiceConfig, error) // Purchase plans diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 2d6ade951..dd276931f 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -399,20 +399,36 @@ func saveGlobalConfigWith(ctx context.Context, q globalConfigExecutor, config *G // SERVICE CONFIGURATION // ========================================== -// GetServiceConfig retrieves configuration for a specific service. -func (s *PostgresStore) GetServiceConfig(ctx context.Context, provider, service string) (*ServiceConfig, error) { - query := ` - SELECT provider, service, enabled, term, payment, coverage, ramp_schedule, +// serviceConfigColumns is the column list every service_configs read scans, +// in scanServiceConfig order. +const serviceConfigColumns = `provider, service, enabled, term, payment, coverage, ramp_schedule, include_engines, exclude_engines, include_regions, exclude_regions, - include_types, exclude_types, min_count + include_types, exclude_types, min_count` + +const getServiceConfigQuery = `SELECT ` + serviceConfigColumns + ` FROM service_configs - WHERE provider = $1 AND service = $2 - ` + WHERE provider = $1 AND service = $2` + +// serviceConfigLockPrefix namespaces the per-service advisory lock +// UpdateServiceConfigAtomic takes. It is a distinct text input to the same +// hashtextextended(text, 0) bigint key space as the ladder-config and +// plan-step provider locks; a 64-bit hash collision could only cause a +// spurious wait, never a correctness break. +const serviceConfigLockPrefix = "cudly:service-config:" + +// GetServiceConfig retrieves configuration for a specific service. +func (s *PostgresStore) GetServiceConfig(ctx context.Context, provider, service string) (*ServiceConfig, error) { + return getServiceConfigFrom(ctx, s.db, getServiceConfigQuery, provider, service) +} +// getServiceConfigFrom runs query (getServiceConfigQuery, optionally with a +// locking clause) via q (the pool or a transaction). A missing row is reported +// as an error wrapping ErrNotFound. +func getServiceConfigFrom(ctx context.Context, q globalConfigExecutor, query, provider, service string) (*ServiceConfig, error) { var config ServiceConfig var includeEngines, excludeEngines, includeRegions, excludeRegions, includeTypes, excludeTypes []string - err := s.db.QueryRow(ctx, query, provider, service).Scan( + err := q.QueryRow(ctx, query, provider, service).Scan( &config.Provider, &config.Service, &config.Enabled, @@ -449,6 +465,59 @@ func (s *PostgresStore) GetServiceConfig(ctx context.Context, provider, service // SaveServiceConfig saves configuration for a service. func (s *PostgresStore) SaveServiceConfig(ctx context.Context, config *ServiceConfig) error { + return saveServiceConfigWith(ctx, s.db, config) +} + +// UpdateServiceConfigAtomic performs a serialized read-merge-write of one +// service_configs row (issue #1488). In one transaction it takes a +// transaction-scoped advisory lock on (provider, service), reads the row +// FOR UPDATE, hands it to apply and upserts the row apply returns. +// +// Both locks are needed. The advisory lock serializes two saves of a row that +// does not exist yet, which a row lock cannot do because there is nothing to +// lock: without it both would read nothing and the second insert would +// overwrite the first. The row lock makes writers that do not take the +// advisory lock, such as the global-defaults propagation in +// UpdateGlobalConfigAtomic, wait for this transaction (and this transaction +// wait for them), so the row read here is the row the write replaces. +// +// apply receives nil when no row exists. It must return the full row to +// write, keyed on the same provider and service; an error it returns aborts +// the transaction and is propagated unchanged. +func (s *PostgresStore) UpdateServiceConfigAtomic(ctx context.Context, provider, service string, apply func(existing *ServiceConfig) (*ServiceConfig, error)) (*ServiceConfig, error) { + var saved *ServiceConfig + err := s.WithTx(ctx, func(tx pgx.Tx) error { + if _, err := tx.Exec(ctx, "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", serviceConfigLockPrefix+provider+"/"+service); err != nil { + return fmt.Errorf("failed to acquire service config lock: %w", err) + } + existing, err := getServiceConfigFrom(ctx, tx, getServiceConfigQuery+" FOR UPDATE", provider, service) + if errors.Is(err, ErrNotFound) { + existing = nil + } else if err != nil { + return err + } + next, err := apply(existing) + if err != nil { + return err + } + if next == nil || next.Provider != provider || next.Service != service { + return fmt.Errorf("service config update for %s/%s must return a row with the same key", provider, service) + } + if err := saveServiceConfigWith(ctx, tx, next); err != nil { + return err + } + saved = next + return nil + }) + if err != nil { + return nil, err + } + return saved, nil +} + +// saveServiceConfigWith upserts one service_configs row via q (the pool or a +// transaction). +func saveServiceConfigWith(ctx context.Context, q globalConfigExecutor, config *ServiceConfig) error { query := ` INSERT INTO service_configs ( provider, service, enabled, term, payment, coverage, ramp_schedule, @@ -471,7 +540,7 @@ func (s *PostgresStore) SaveServiceConfig(ctx context.Context, config *ServiceCo updated_at = NOW() ` - _, err := s.db.Exec(ctx, query, + _, err := q.Exec(ctx, query, config.Provider, config.Service, config.Enabled, diff --git a/internal/config/store_postgres_service_config_atomic_integration_test.go b/internal/config/store_postgres_service_config_atomic_integration_test.go new file mode 100644 index 000000000..c94cf5fbe --- /dev/null +++ b/internal/config/store_postgres_service_config_atomic_integration_test.go @@ -0,0 +1,107 @@ +//go:build integration + +package config + +import ( + "context" + "errors" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/require" +) + +// Deterministic lock tests for UpdateServiceConfigAtomic (issue #1488). The +// handler-level race tests in internal/api fire real concurrent PUTs; these +// pin each of the two locks the store method takes by parking a competing +// writer mid-transaction and checking the update queues behind it and then +// merges onto what that writer committed. + +// holdServiceConfigWriter opens a transaction for the test to play a competing +// writer in. It rolls back at test end unless the test committed it. +func holdServiceConfigWriter(ctx context.Context, t *testing.T, store *PostgresStore) pgx.Tx { + t.Helper() + tx, err := store.db.Begin(ctx) + require.NoError(t, err) + t.Cleanup(func() { + if rbErr := tx.Rollback(ctx); rbErr != nil && !errors.Is(rbErr, pgx.ErrTxClosed) { + t.Logf("writer rollback: %v", rbErr) + } + }) + return tx +} + +// setCoverage80 is the update under test: a partial PUT of coverage only. +func setCoverage80(existing *ServiceConfig) (*ServiceConfig, error) { + next := &ServiceConfig{Provider: "aws", Service: "rds", Term: 1, Payment: "no-upfront"} + if existing != nil { + copied := *existing + next = &copied + } + next.Coverage = 80 + return next, nil +} + +// TestUpdateServiceConfigAtomic_QueuesBehindRowWriter: the global-defaults +// propagation rewrites service rows with a plain UPDATE, without the +// per-service advisory lock. The FOR UPDATE read must wait for it and merge +// onto its committed term; reading the row unlocked would write the old term +// back over it. +func TestUpdateServiceConfigAtomic_QueuesBehindRowWriter(t *testing.T) { + ctx := context.Background() + store := NewPostgresStore(testhelpers.RequireMigratedDatabase(ctx, t, getTestMigrationsPath()).DB) + require.NoError(t, store.SaveServiceConfig(ctx, &ServiceConfig{ + Provider: "aws", Service: "rds", Enabled: true, Term: 1, Payment: "no-upfront", Coverage: 50, RampSchedule: RampImmediate, + })) + + writer := holdServiceConfigWriter(ctx, t, store) + _, err := writer.Exec(ctx, `UPDATE service_configs SET term = 3 WHERE provider = 'aws' AND service = 'rds'`) + require.NoError(t, err) + + done := make(chan error, 1) + go func() { + _, uErr := store.UpdateServiceConfigAtomic(ctx, "aws", "rds", setCoverage80) + done <- uErr + }() + testhelpers.WaitForLockWait(ctx, t, store.db, "%service_configs%") + require.NoError(t, writer.Commit(ctx)) + require.NoError(t, testhelpers.Await(t, done)) + + got, err := store.GetServiceConfig(ctx, "aws", "rds") + require.NoError(t, err) + require.Equal(t, 3, got.Term, "the concurrent writer's term must survive") + require.Equal(t, 80.0, got.Coverage) +} + +// TestUpdateServiceConfigAtomic_QueuesBehindFirstInsert: when no row exists +// there is nothing to lock FOR UPDATE, so only the advisory lock stops two +// first saves from both reading nothing. A competing save holding that lock +// inserts the row; the update must wait, then merge onto the inserted row +// rather than overwrite it with its own fresh one. +func TestUpdateServiceConfigAtomic_QueuesBehindFirstInsert(t *testing.T) { + ctx := context.Background() + store := NewPostgresStore(testhelpers.RequireMigratedDatabase(ctx, t, getTestMigrationsPath()).DB) + + writer := holdServiceConfigWriter(ctx, t, store) + _, err := writer.Exec(ctx, "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", serviceConfigLockPrefix+"aws/rds") + require.NoError(t, err) + require.NoError(t, saveServiceConfigWith(ctx, writer, &ServiceConfig{ + Provider: "aws", Service: "rds", Enabled: true, Term: 1, Payment: "no-upfront", Coverage: 50, RampSchedule: RampImmediate, MinCount: 5, + })) + + done := make(chan error, 1) + go func() { + _, uErr := store.UpdateServiceConfigAtomic(ctx, "aws", "rds", setCoverage80) + done <- uErr + }() + testhelpers.WaitForLockWait(ctx, t, store.db, "") + require.NoError(t, writer.Commit(ctx)) + require.NoError(t, testhelpers.Await(t, done)) + + got, err := store.GetServiceConfig(ctx, "aws", "rds") + require.NoError(t, err) + require.Equal(t, 5, got.MinCount, "the first save's min_count must survive") + require.Equal(t, 80.0, got.Coverage) + require.True(t, got.Enabled) +} diff --git a/internal/config/store_postgres_service_config_atomic_test.go b/internal/config/store_postgres_service_config_atomic_test.go new file mode 100644 index 000000000..868d39f10 --- /dev/null +++ b/internal/config/store_postgres_service_config_atomic_test.go @@ -0,0 +1,131 @@ +package config + +import ( + "context" + "errors" + "testing" + + "github.com/pashagolub/pgxmock/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var serviceConfigCols = []string{ + "provider", "service", "enabled", "term", "payment", "coverage", "ramp_schedule", + "include_engines", "exclude_engines", "include_regions", "exclude_regions", + "include_types", "exclude_types", "min_count", +} + +// expectServiceConfigLockedRead queues the BEGIN, the per-service advisory +// lock and the FOR UPDATE read UpdateServiceConfigAtomic issues, in order. +func expectServiceConfigLockedRead(mock pgxmock.PgxPoolIface, rows *pgxmock.Rows) { + mock.ExpectBegin() + mock.ExpectExec("pg_advisory_xact_lock").WithArgs(serviceConfigLockPrefix + "aws/rds"). + WillReturnResult(pgxmock.NewResult("SELECT", 1)) + mock.ExpectQuery(`FROM service_configs\s+WHERE provider = \$1 AND service = \$2 FOR UPDATE`). + WithArgs("aws", "rds").WillReturnRows(rows) +} + +// TestPGXMock_UpdateServiceConfigAtomic_LockedReadMergeWrite pins issue #1488: +// the read, the caller's merge and the upsert run in ONE transaction, after the +// per-service advisory lock and with the row read FOR UPDATE. pgxmock enforces +// expectation order, so a pass means no statement escaped the transaction. +func TestPGXMock_UpdateServiceConfigAtomic_LockedReadMergeWrite(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + + expectServiceConfigLockedRead(mock, pgxmock.NewRows(serviceConfigCols).AddRow( + "aws", "rds", true, 3, "all-upfront", 50.0, RampImmediate, + []string{"mysql"}, []string{}, []string{}, []string{}, []string{}, []string{}, 4, + )) + mock.ExpectExec("INSERT INTO service_configs"). + WithArgs("aws", "rds", true, 3, "all-upfront", 80.0, RampImmediate, + []string{"mysql"}, []string{}, []string{}, []string{}, []string{}, []string{}, 4). + WillReturnResult(pgxmock.NewResult("INSERT", 1)) + mock.ExpectCommit() + + saved, err := store.UpdateServiceConfigAtomic(context.Background(), "aws", "rds", func(existing *ServiceConfig) (*ServiceConfig, error) { + require.NotNil(t, existing, "apply must see the locked row") + next := *existing + next.Coverage = 80 + return &next, nil + }) + require.NoError(t, err) + assert.Equal(t, 80.0, saved.Coverage) + assert.Equal(t, 4, saved.MinCount) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +// TestPGXMock_UpdateServiceConfigAtomic_NoRowPassesNil: with no stored row, +// apply gets nil (not an error) and its row is inserted under the same lock. +func TestPGXMock_UpdateServiceConfigAtomic_NoRowPassesNil(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + + expectServiceConfigLockedRead(mock, pgxmock.NewRows(serviceConfigCols)) + mock.ExpectExec("INSERT INTO service_configs").WithArgs(anyArgsCfg(14)...). + WillReturnResult(pgxmock.NewResult("INSERT", 1)) + mock.ExpectCommit() + + sawNil := false + _, err := store.UpdateServiceConfigAtomic(context.Background(), "aws", "rds", func(existing *ServiceConfig) (*ServiceConfig, error) { + sawNil = existing == nil + return &ServiceConfig{Provider: "aws", Service: "rds", Term: 1, Payment: "no-upfront"}, nil + }) + require.NoError(t, err) + assert.True(t, sawNil) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +// TestPGXMock_UpdateServiceConfigAtomic_AbortsWithoutWrite: an apply error (a +// validation failure) or a row keyed on another service rolls back with no +// upsert. The apply error propagates unchanged. +func TestPGXMock_UpdateServiceConfigAtomic_AbortsWithoutWrite(t *testing.T) { + sentinel := errors.New("validation failed") + cases := map[string]struct { + next *ServiceConfig + err error + wantErr string + }{ + "apply error": {err: sentinel, wantErr: "validation failed"}, + "nil row": {wantErr: "same key"}, + "other service": {next: &ServiceConfig{Provider: "aws", Service: "ec2"}, wantErr: "same key"}, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + expectServiceConfigLockedRead(mock, pgxmock.NewRows(serviceConfigCols)) + // No ExpectExec("INSERT ..."): the upsert must not run. + mock.ExpectRollback() + + _, err := store.UpdateServiceConfigAtomic(context.Background(), "aws", "rds", func(*ServiceConfig) (*ServiceConfig, error) { + return tc.next, tc.err + }) + require.ErrorContains(t, err, tc.wantErr) + if tc.err != nil { + assert.ErrorIs(t, err, tc.err) + } + assert.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +// TestPGXMock_UpdateServiceConfigAtomic_ReadErrorAborts: a failed read is not +// mistaken for a missing row; apply never runs and nothing is written. +func TestPGXMock_UpdateServiceConfigAtomic_ReadErrorAborts(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + mock.ExpectBegin() + mock.ExpectExec("pg_advisory_xact_lock").WithArgs(pgxmock.AnyArg()). + WillReturnResult(pgxmock.NewResult("SELECT", 1)) + mock.ExpectQuery("FOR UPDATE").WithArgs("aws", "rds").WillReturnError(errors.New("connection reset")) + mock.ExpectRollback() + + _, err := store.UpdateServiceConfigAtomic(context.Background(), "aws", "rds", func(*ServiceConfig) (*ServiceConfig, error) { + t.Fatal("apply must not run after a failed read") + return nil, nil + }) + require.ErrorContains(t, err, "connection reset") + assert.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 9f42ece62..497125a1e 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -157,6 +157,49 @@ func (m *MockConfigStore) SaveServiceConfig(ctx context.Context, cfg *config.Ser return args.Error(0) } +// UpdateServiceConfigAtomic mocks the locked read-merge-write of one service +// config row. When an expectation is registered it dispatches through +// m.Called (tests can drive the apply closure via a Run callback). Otherwise +// it emulates the real store by routing through the mocked GetServiceConfig + +// apply + SaveServiceConfig, so tests that seed those two keep working. Like +// the real store, a not-found read hands apply a nil row. +func (m *MockConfigStore) UpdateServiceConfigAtomic(ctx context.Context, provider, service string, apply func(*config.ServiceConfig) (*config.ServiceConfig, error)) (*config.ServiceConfig, error) { + m.record("UpdateServiceConfigAtomic", ctx, provider, service, apply) + if isExpected(&m.Mock, "UpdateServiceConfigAtomic") { + args := m.Called(ctx, provider, service, apply) + if args.Get(0) == nil { + return nil, args.Error(1) + } + v, ok := args.Get(0).(*config.ServiceConfig) + if !ok { + m.reporter.wrongReturnType("UpdateServiceConfigAtomic", "*config.ServiceConfig", args.Get(0)) + } + return v, args.Error(1) + } + return m.updateServiceConfigFallback(ctx, provider, service, apply) +} + +// updateServiceConfigFallback is UpdateServiceConfigAtomic's default. It +// routes through the mocked GetServiceConfig and SaveServiceConfig, which +// record themselves; see UpdatePurchasePlanTx. +func (m *MockConfigStore) updateServiceConfigFallback(ctx context.Context, provider, service string, apply func(*config.ServiceConfig) (*config.ServiceConfig, error)) (*config.ServiceConfig, error) { + existing, err := m.GetServiceConfig(ctx, provider, service) + if err != nil && !errors.Is(err, config.ErrNotFound) { + return nil, err + } + next, err := apply(existing) + if err != nil { + return nil, err + } + if next == nil || next.Provider != provider || next.Service != service { + return nil, fmt.Errorf("service config update for %s/%s must return a row with the same key", provider, service) + } + if err := m.SaveServiceConfig(ctx, next); err != nil { + return nil, err + } + return next, nil +} + // ListServiceConfigs mocks the ListServiceConfigs operation. func (m *MockConfigStore) ListServiceConfigs(ctx context.Context) ([]config.ServiceConfig, error) { m.record("ListServiceConfigs", ctx)