From 54c4ff48fa3937e512d79f6f676d16c88f831809 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Mon, 26 Oct 2026 19:31:32 +0000 Subject: [PATCH 1/2] fix(config): make the service-config PUT read-merge-write atomic The service-config PUT read the stored row, merged the request's present keys into it and upserted the full row, with no transaction or lock. Two admins saving different fields of the same row at once both merged onto the same base and the later write dropped the earlier change. Add PostgresStore.UpdateServiceConfigAtomic, mirroring UpdateGlobalConfigAtomic: one transaction takes a per-(provider, service) advisory lock (serializes two first saves of a row that does not exist yet), reads the row FOR UPDATE (so the global-defaults propagation, which updates service rows without the advisory lock, cannot be overwritten by a stale read), runs the caller's merge and validation, and upserts. The handler now merges and validates inside that callback; mergeServiceConfig becomes a pure overlay onto the row the store hands it. The mock gets a matching UpdateServiceConfigAtomic that falls back to the mocked Get/Save. Real-Postgres tests fire 40 rounds of two concurrent PUTs of different keys (existing row and first save); before the fix 39 of 40 rounds lost an update in each. Store-level lock tests pin each of the two locks. Closes #1488 Co-Authored-By: Claude Opus 5.5 --- internal/api/coverage_gaps_test.go | 56 ++------ internal/api/handler_config.go | 80 +++++------ ...er_config_service_race_integration_test.go | 129 +++++++++++++++++ internal/api/handler_config_test.go | 111 +++++++++++---- internal/config/interfaces.go | 6 + internal/config/store_postgres.go | 87 ++++++++++-- ..._service_config_atomic_integration_test.go | 107 ++++++++++++++ ...ore_postgres_service_config_atomic_test.go | 131 ++++++++++++++++++ internal/mocks/stores.go | 43 ++++++ 9 files changed, 634 insertions(+), 116 deletions(-) create mode 100644 internal/api/handler_config_service_race_integration_test.go create mode 100644 internal/config/store_postgres_service_config_atomic_integration_test.go create mode 100644 internal/config/store_postgres_service_config_atomic_test.go 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..915b49833 100644 --- a/internal/api/handler_config.go +++ b/internal/api/handler_config.go @@ -203,36 +203,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 +353,37 @@ 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 + // 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 h.mergeAndValidateServiceConfig(ctx, existing, cfg, req.Body) + }) + if err != nil { + return nil, err } - cfg = merged - // Validate the configuration - if err := cfg.Validate(); err != nil { - return nil, NewClientError(400, fmt.Sprintf("validation error: %s", err)) - } + return &StatusResponse{Status: "updated"}, nil +} - if err := h.checkCommitmentOptionCombo(ctx, cfg); err != 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. +func (h *Handler) mergeAndValidateServiceConfig(ctx context.Context, existing *config.ServiceConfig, cfg config.ServiceConfig, body string) (*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 := h.checkCommitmentOptionCombo(ctx, 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..03db55eda --- /dev/null +++ b/internal/api/handler_config_service_race_integration_test.go @@ -0,0 +1,129 @@ +//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" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/aws/aws-lambda-go/events" + "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) +} diff --git a/internal/api/handler_config_test.go b/internal/api/handler_config_test.go index c89d073de..b2129d36e 100644 --- a/internal/api/handler_config_test.go +++ b/internal/api/handler_config_test.go @@ -645,11 +645,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 +658,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 +666,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 +678,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 +693,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 +700,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 +711,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 +719,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 +809,86 @@ 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_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) diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 4046e1969..f20871f20 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) From 8fcb1af082f1aa91ae6d36665dc547d8398adc28 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Mon, 26 Oct 2026 21:13:08 +0000 Subject: [PATCH 2/2] fix(api): check commitment options outside the service-config transaction The service-config PUT now runs its read-merge-write in one locked transaction (#1488), and the merge callback ran the commitment-option check, whose Validate reads the commitmentopts store (and, on a cold cache, lists cloud accounts and probes AWS) through the same pool. Each save then held one pooled connection and waited for a second. The Settings page saves its AWS rows in parallel, so with as many saves as pool connections (3 on Fargate, 6 on Cloud Run) every save waited on the others until its deadline and the instance starved. Fetch the options with Get before the transaction opens and check the merged row inside it with the new pure commitmentopts.ValidateCombo, which Service.Validate now also uses. A read error is kept and only fails a save whose merged row carries a term and payment, as before. The rest of the callback (merge, ServiceConfig.Validate) was already store-free; the store side runs every statement on the transaction. The other combo-check callers (account override PUT, override import planning) do not hold a transaction and keep using Validate. Adds a real-Postgres test with a pool of 3 and the nine Settings AWS rows saved at once, wired as app.go wires the service: it timed out on every save before this change and passes in about 100ms after it. Co-Authored-By: Claude Opus 5.5 --- internal/api/handler_config.go | 83 ++++++++++++++-- ...er_config_service_race_integration_test.go | 76 +++++++++++++++ internal/api/handler_config_test.go | 94 ++++++++++++++++--- internal/commitmentopts/service.go | 22 ++++- internal/commitmentopts/service_test.go | 16 ++++ 5 files changed, 266 insertions(+), 25 deletions(-) diff --git a/internal/api/handler_config.go b/internal/api/handler_config.go index 915b49833..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 } @@ -353,13 +409,18 @@ func (h *Handler) updateServiceConfig(ctx context.Context, req *events.LambdaFun cfg.Service = parts[1] } + // 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) + // 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 h.mergeAndValidateServiceConfig(ctx, existing, cfg, req.Body) + return mergeAndValidateServiceConfig(existing, cfg, req.Body, combos) }) if err != nil { return nil, err @@ -373,8 +434,10 @@ func (h *Handler) updateServiceConfig(ctx context.Context, req *events.LambdaFun // 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. -func (h *Handler) mergeAndValidateServiceConfig(ctx context.Context, existing *config.ServiceConfig, cfg config.ServiceConfig, body string) (*config.ServiceConfig, error) { +// 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 @@ -382,7 +445,7 @@ func (h *Handler) mergeAndValidateServiceConfig(ctx context.Context, existing *c if err := merged.Validate(); err != nil { return nil, NewClientError(400, fmt.Sprintf("validation error: %s", err)) } - if err := h.checkCommitmentOptionCombo(ctx, merged); err != nil { + if err := combos.check(merged); err != nil { return nil, err } 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 index 03db55eda..a2b20aa5c 100644 --- a/internal/api/handler_config_service_race_integration_test.go +++ b/internal/api/handler_config_service_race_integration_test.go @@ -14,10 +14,14 @@ import ( "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" ) @@ -127,3 +131,75 @@ func TestUpdateServiceConfig_ConcurrentFirstSavesKeepBothKeys(t *testing.T) { } 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 b2129d36e..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" @@ -847,6 +848,73 @@ func TestHandler_updateServiceConfig_MergesInsideAtomicUpdate(t *testing.T) { 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. @@ -930,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 }, }, } @@ -979,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 }, }, } @@ -1013,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}