From 365e959655c75bbf657a754d5a56ec0a7f377db0 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 23 Oct 2026 04:58:30 +0000 Subject: [PATCH] fix(purchase,commitmentopts): close out the open #243 review findings - purchase: executableByScheduler skips a direct-execute row (PlanID "") before the store instead of counting a 22P02 failure every tick, and refuses a (nil, nil) plan instead of dereferencing it (A05-013). - purchase: FinalizeInFlightRevocations waits out its retry backoff with a ctx-aware timer and stops the sweep, returning the partial counts, when ctx is done. Adds the missing table test for its accounting (A16-005). - purchase: correct the stale Azure savings-plans re-drive comments; the refusal itself is unchanged. - commitmentopts: a page-capped OpenSearch/Redshift probe records no combos for the service, so Validate stays permissive instead of rejecting combos that fell past the cap. - commitmentopts: findAWSAccount asks for enabled accounts only (A05-010). - commitmentopts: the Azure SP probe needs a Valid=true benefit to record a combo as sold (A05-012). - api: a commitment-option Validate error other than ErrNoData fails the service-config save with a 503 instead of letting it through (A02-022). Refs #243 Co-Authored-By: Claude Opus 5.5 --- internal/api/handler_config.go | 19 +++- internal/api/handler_config_test.go | 38 +++++++ internal/commitmentopts/probe.go | 47 ++++++-- internal/commitmentopts/probe_azure.go | 24 ++-- internal/commitmentopts/probe_azure_test.go | 35 +++++- internal/commitmentopts/probe_test.go | 56 ++++++++++ internal/commitmentopts/service.go | 12 +- internal/commitmentopts/service_test.go | 65 +++++++++++ internal/purchase/finalize_revocations.go | 44 ++++++-- .../purchase/finalize_revocations_test.go | 103 ++++++++++++++++++ internal/purchase/manager.go | 64 ++++++++--- .../purchase/money_path_regression_test.go | 1 + internal/purchase/reaper.go | 6 +- internal/purchase/step_cap_test.go | 24 ++++ 14 files changed, 481 insertions(+), 57 deletions(-) create mode 100644 internal/purchase/finalize_revocations_test.go diff --git a/internal/api/handler_config.go b/internal/api/handler_config.go index 74dcdb067..18a57957a 100644 --- a/internal/api/handler_config.go +++ b/internal/api/handler_config.go @@ -4,12 +4,14 @@ package api import ( "context" "encoding/json" + "errors" "fmt" "os" "strings" "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/commitmentopts" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" "github.com/aws/aws-lambda-go/events" ) @@ -158,18 +160,25 @@ func (h *Handler) requireGlobalConfigScope(ctx context.Context, session *Session // combination we've dynamically confirmed the cloud doesn't sell. Returns // nil when: no probe service is wired, the service hasn't persisted data // yet (absent data → fall through to the frontend's hardcoded rules), -// the save isn't AWS, or the combo is valid. Errors from Validate are -// logged and swallowed (permissive) so a transient DB blip never blocks -// a settings save. +// the save isn't AWS, or the combo is valid. +// +// Any other Validate error (a store read failing, a timeout) fails the save +// with a 503 instead of letting it through: the check exists for API +// callers that bypass the frontend's rules, so waving the save through on an +// error would persist an unsupported combo that fails only at purchase time +// (issue #243, audit finding A02-022). The caller can retry. func (h *Handler) checkCommitmentOptionCombo(ctx context.Context, cfg config.ServiceConfig) error { if h.commitmentOpts == nil || cfg.Provider != "aws" || cfg.Term <= 0 || cfg.Payment == "" { return nil } ok, err := h.commitmentOpts.Validate(ctx, cfg.Provider, cfg.Service, cfg.Term, cfg.Payment) - if err != nil { - logging.Warnf("commitment-option validation error (allowing save): %v", err) + if errors.Is(err, commitmentopts.ErrNoData) { 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") + } if !ok { return NewClientError(400, fmt.Sprintf( "%s does not support %dyr %s commitments", diff --git a/internal/api/handler_config_test.go b/internal/api/handler_config_test.go index 44c388030..924a8185e 100644 --- a/internal/api/handler_config_test.go +++ b/internal/api/handler_config_test.go @@ -2,6 +2,7 @@ package api import ( "context" + "errors" "fmt" "sync" "testing" @@ -827,6 +828,43 @@ func TestHandler_updateServiceConfig_CommitmentOptsAccept(t *testing.T) { assert.Equal(t, "updated", result.Status) } +// A Validate error other than ErrNoData must fail the save closed with a 503 +// rather than persist a combo nobody checked (issue #243, A02-022). +func TestHandler_updateServiceConfig_CommitmentOptsErrorRefusesSave(t *testing.T) { + ctx := context.Background() + mockStore := new(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() + mockStore.On("GetServiceConfig", ctx, "aws", "rds").Return(nil, nil) + + handler := &Handler{ + 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") + }, + }, + } + + req := &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"Authorization": "Bearer admin-token"}, + Body: `{"enabled": true, "term": 3, "payment": "no-upfront", "coverage": 80}`, + } + result, err := handler.updateServiceConfig(ctx, req, "aws/rds") + + require.Error(t, err) + assert.Nil(t, result) + ce, ok := IsClientError(err) + require.True(t, ok) + assert.Equal(t, 503, ce.code) + mockStore.AssertNotCalled(t, "SaveServiceConfig", mock.Anything, mock.Anything) +} + func TestHandler_updateServiceConfig_NoSlash(t *testing.T) { ctx := context.Background() mockStore := new(MockConfigStore) diff --git a/internal/commitmentopts/probe.go b/internal/commitmentopts/probe.go index 2a0cc5e35..60660ed41 100644 --- a/internal/commitmentopts/probe.go +++ b/internal/commitmentopts/probe.go @@ -4,6 +4,7 @@ import ( "context" "fmt" + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/ec2" ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types" @@ -81,20 +82,46 @@ func walkPaginated( service string, fetchPage func(ctx context.Context, token *string) ([]rawOffer, *string, error), ) ([]rawOffer, error) { - var raw []rawOffer + raw, _, err := walkPaginatedCapped(ctx, service, fetchPage) + return raw, err +} + +// walkPaginatedCapped is walkPaginated that also reports whether the page cap +// stopped the walk while AWS still had pages to return (capped=true), as +// opposed to pagination running out. +func walkPaginatedCapped( + ctx context.Context, + service string, + fetchPage func(ctx context.Context, token *string) ([]rawOffer, *string, error), +) (raw []rawOffer, capped bool, err error) { var token *string for page := 0; page < maxPages; page++ { - offers, next, err := fetchPage(ctx, token) - if err != nil { - return nil, fmt.Errorf("%s: %w", service, err) + offers, next, fetchErr := fetchPage(ctx, token) + if fetchErr != nil { + return nil, false, fmt.Errorf("%s: %w", service, fetchErr) } raw = append(raw, offers...) if next == nil || aws.ToString(next) == "" { - break + return raw, false, nil } token = next } - return raw, nil + return raw, true, nil +} + +// collectUnlessCapped is collect for the probers that filter the probe target +// client-side (OpenSearch, Redshift): their APIs take no instance-type filter, +// so the walk pages through every type and the page cap can cut off part of +// the target's (term, payment) matrix. A capped walk yields no combos for the +// service, which Service.Validate treats like an unprobed service +// (permissive, with a warning), instead of a partial list that would reject +// combos AWS does sell (issue #243). +func collectUnlessCapped(service string, raw []rawOffer, capped bool) []Combo { + if capped { + logging.Warnf("commitmentopts: %s probe hit the %d-page cap before pagination ended; recording no combos for it so validation stays permissive", service, maxPages) + return nil + } + return collect(service, raw) } // collect dedupes a probe's raw (durationSeconds, rawPayment) pairs, runs @@ -270,7 +297,7 @@ func (p *OpenSearchProber) Service() string { return "opensearch" } // Probe returns the combos for t3.small.search. func (p *OpenSearchProber) Probe(ctx context.Context, cfg aws.Config) ([]Combo, error) { client := p.client(cfg) - raw, err := walkPaginated(ctx, p.Service(), func(ctx context.Context, token *string) ([]rawOffer, *string, error) { + raw, capped, err := walkPaginatedCapped(ctx, p.Service(), func(ctx context.Context, token *string) ([]rawOffer, *string, error) { out, err := client.DescribeReservedInstanceOfferings(ctx, &opensearch.DescribeReservedInstanceOfferingsInput{ MaxResults: pageSize, NextToken: token, @@ -293,7 +320,7 @@ func (p *OpenSearchProber) Probe(ctx context.Context, cfg aws.Config) ([]Combo, if err != nil { return nil, err } - return collect(p.Service(), raw), nil + return collectUnlessCapped(p.Service(), raw, capped), nil } func (p *OpenSearchProber) client(cfg aws.Config) OpenSearchDescribeOfferings { @@ -325,7 +352,7 @@ func (p *RedshiftProber) Service() string { return "redshift" } // Probe returns the combos for dc2.large. func (p *RedshiftProber) Probe(ctx context.Context, cfg aws.Config) ([]Combo, error) { client := p.client(cfg) - raw, err := walkPaginated(ctx, p.Service(), func(ctx context.Context, token *string) ([]rawOffer, *string, error) { + raw, capped, err := walkPaginatedCapped(ctx, p.Service(), func(ctx context.Context, token *string) ([]rawOffer, *string, error) { out, err := client.DescribeReservedNodeOfferings(ctx, &redshift.DescribeReservedNodeOfferingsInput{ MaxRecords: aws.Int32(pageSize), Marker: token, @@ -348,7 +375,7 @@ func (p *RedshiftProber) Probe(ctx context.Context, cfg aws.Config) ([]Combo, er if err != nil { return nil, err } - return collect(p.Service(), raw), nil + return collectUnlessCapped(p.Service(), raw, capped), nil } func (p *RedshiftProber) client(cfg aws.Config) RedshiftDescribeOfferings { diff --git a/internal/commitmentopts/probe_azure.go b/internal/commitmentopts/probe_azure.go index 2ebddddd6..307a3df86 100644 --- a/internal/commitmentopts/probe_azure.go +++ b/internal/commitmentopts/probe_azure.go @@ -163,9 +163,10 @@ func azureProbeCurrency(code string) (string, error) { // probeCombo calls ValidatePurchase for a single (term, billingPlan) pair and // returns true if the API considers it a valid offering. // -// A response where any benefit has Valid=false is treated as "not offered" -// (returns false, nil). Any other API error is returned as-is so the caller -// can decide whether it is a transient failure or a configuration problem. +// A response where any benefit has Valid=false, or where no benefit carries +// Valid=true, is treated as "not offered" (returns false, nil). Any other API +// error is returned as-is so the caller can decide whether it is a transient +// failure or a configuration problem. func (p *AzureSPProber) probeCombo( ctx context.Context, client AzureSPValidateAPI, @@ -210,15 +211,22 @@ func (p *AzureSPProber) probeCombo( return false, err } - // The API returns a per-benefit valid flag. If all reported benefits are - // valid (or the slice is empty, which can't happen for a single-item - // request in practice), the combo is available. + // The API returns a per-benefit valid flag. The combo counts as sold only + // on positive confirmation: at least one benefit with Valid=true and none + // with Valid=false. An empty Benefits slice or a nil flag says nothing, so + // it must not record a combo (such as a P5Y plan) as purchasable + // (issue #243, audit finding A05-012). + confirmed := false for _, b := range resp.Benefits { - if b != nil && b.Valid != nil && !*b.Valid { + if b == nil || b.Valid == nil { + continue + } + if !*b.Valid { return false, nil } + confirmed = true } - return true, nil + return confirmed, nil } func (p *AzureSPProber) client(cred azcore.TokenCredential) (AzureSPValidateAPI, error) { diff --git a/internal/commitmentopts/probe_azure_test.go b/internal/commitmentopts/probe_azure_test.go index d85c5aa39..1210b35b0 100644 --- a/internal/commitmentopts/probe_azure_test.go +++ b/internal/commitmentopts/probe_azure_test.go @@ -136,19 +136,20 @@ func TestAzureSPProber_NoneValid(t *testing.T) { } func TestAzureSPProber_EmptyBenefitsSlice(t *testing.T) { - // An API response with an empty benefits slice counts as "valid" - // (no invalid flag returned) so the combo is included. + // An empty benefits slice confirms nothing, so no combo is recorded as + // sold: the probe requires positive confirmation (audit finding A05-012). p := newFakeProber(func(_ armbillingbenefits.SavingsPlanPurchaseValidateRequest) (armbillingbenefits.RPClientValidatePurchaseResponse, error) { return armbillingbenefits.RPClientValidatePurchaseResponse{}, nil }) got, err := p.ProbeAzure(context.Background(), nil, "USD") require.NoError(t, err) - assert.Len(t, got, 6, "empty benefits slice treated as valid") + assert.Empty(t, got, "an empty benefits slice must not record a combo as sold") } func TestAzureSPProber_NilValidFlag(t *testing.T) { - // A benefit with a nil Valid pointer is not invalid — treated as valid. + // A benefit with a nil Valid pointer confirms nothing, so the combo is + // not recorded as sold (audit finding A05-012). p := newFakeProber(func(_ armbillingbenefits.SavingsPlanPurchaseValidateRequest) (armbillingbenefits.RPClientValidatePurchaseResponse, error) { return armbillingbenefits.RPClientValidatePurchaseResponse{ SavingsPlanValidateResponse: armbillingbenefits.SavingsPlanValidateResponse{ @@ -161,7 +162,31 @@ func TestAzureSPProber_NilValidFlag(t *testing.T) { got, err := p.ProbeAzure(context.Background(), nil, "USD") require.NoError(t, err) - assert.Len(t, got, 6, "nil Valid flag treated as valid") + assert.Empty(t, got, "a nil Valid flag must not record a combo as sold") +} + +func TestAzureSPProber_ConfirmationRule(t *testing.T) { + yes, no := true, false + cases := []struct { + name string + benefits []*armbillingbenefits.SavingsPlanValidResponseProperty + want int + }{ + {"nil flag beside a confirmed benefit", []*armbillingbenefits.SavingsPlanValidResponseProperty{nil, {Valid: nil}, {Valid: &yes}}, 6}, + {"any invalid benefit vetoes", []*armbillingbenefits.SavingsPlanValidResponseProperty{{Valid: &yes}, {Valid: &no}}, 0}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + p := newFakeProber(func(armbillingbenefits.SavingsPlanPurchaseValidateRequest) (armbillingbenefits.RPClientValidatePurchaseResponse, error) { + return armbillingbenefits.RPClientValidatePurchaseResponse{ + SavingsPlanValidateResponse: armbillingbenefits.SavingsPlanValidateResponse{Benefits: tc.benefits}, + }, nil + }) + got, err := p.ProbeAzure(context.Background(), nil, "USD") + require.NoError(t, err) + assert.Len(t, got, tc.want) + }) + } } // --------------------------------------------------------------------------- diff --git a/internal/commitmentopts/probe_test.go b/internal/commitmentopts/probe_test.go index 48dbda377..25585ae73 100644 --- a/internal/commitmentopts/probe_test.go +++ b/internal/commitmentopts/probe_test.go @@ -664,3 +664,59 @@ func TestDefaultProbers_IncludesSavingsPlans(t *testing.T) { } assert.True(t, found, "DefaultProbers must include SavingsPlansProber") } + +// --------------------------------------------------------------------------- +// Page cap on the client-side-filtered probers (issue #243) +// --------------------------------------------------------------------------- + +func TestWalkPaginatedCapped_ReportsWhyTheWalkStopped(t *testing.T) { + _, capped, err := walkPaginatedCapped(context.Background(), "test", func(context.Context, *string) ([]rawOffer, *string, error) { + return nil, aws.String("more"), nil + }) + require.NoError(t, err) + assert.True(t, capped, "pages left when the cap fired") + + _, capped, err = walkPaginatedCapped(context.Background(), "test", func(context.Context, *string) ([]rawOffer, *string, error) { + return nil, nil, nil + }) + require.NoError(t, err) + assert.False(t, capped, "pagination ran out") +} + +// OpenSearch and Redshift page through every instance type and filter for the +// probe target client-side, so a walk the cap cut short holds only part of the +// target's matrix. It must yield no combos (Validate then stays permissive for +// the service) rather than a partial list that rejects combos AWS does sell. +func TestFilteredProbers_PageCapYieldsNoCombos(t *testing.T) { + osFake := &fakeOpenSearch{fn: func(*opensearch.DescribeReservedInstanceOfferingsInput) (*opensearch.DescribeReservedInstanceOfferingsOutput, error) { + return &opensearch.DescribeReservedInstanceOfferingsOutput{ + ReservedInstanceOfferings: []opensearchtypes.ReservedInstanceOffering{{ + InstanceType: opensearchtypes.OpenSearchPartitionInstanceType(probeTargetOpenSearch), + Duration: 31536000, + PaymentOption: opensearchtypes.ReservedInstancePaymentOptionNoUpfront, + }}, + NextToken: aws.String("more"), + }, nil + }} + rsFake := &fakeRedshift{fn: func(*redshift.DescribeReservedNodeOfferingsInput) (*redshift.DescribeReservedNodeOfferingsOutput, error) { + return &redshift.DescribeReservedNodeOfferingsOutput{ + ReservedNodeOfferings: []redshifttypes.ReservedNodeOffering{{ + NodeType: aws.String(probeTargetRedshift), + Duration: aws.Int32(31536000), + OfferingType: aws.String("No Upfront"), + }}, + Marker: aws.String("more"), + }, nil + }} + probers := []Prober{ + &OpenSearchProber{NewClient: func(aws.Config) OpenSearchDescribeOfferings { return osFake }}, + &RedshiftProber{NewClient: func(aws.Config) RedshiftDescribeOfferings { return rsFake }}, + } + for _, p := range probers { + t.Run(p.Service(), func(t *testing.T) { + got, err := p.Probe(context.Background(), aws.Config{}) + require.NoError(t, err) + assert.Empty(t, got, "a capped walk must not record a partial matrix") + }) + } +} diff --git a/internal/commitmentopts/service.go b/internal/commitmentopts/service.go index 3931c5090..60bbf8005 100644 --- a/internal/commitmentopts/service.go +++ b/internal/commitmentopts/service.go @@ -125,14 +125,22 @@ func (s *Service) probeAndPersist(ctx context.Context) (Options, error) { // findAWSAccount returns the first enabled AWS account, or nil if none // exist. Nil + nil error is the "no AWS connected" signal. +// +// The filter asks for enabled accounts only, as Scheduler.enabledAccounts +// does: a disabled (for example decommissioned) account listed first would +// otherwise be probed with revoked credentials, and the failed probe would +// leave validation permissive while a healthy account sat later in the list +// (issue #243). The loop re-checks both fields so an AccountLister that +// ignores the filter cannot reintroduce that. func (s *Service) findAWSAccount(ctx context.Context) (*config.CloudAccount, error) { provider := "aws" - accounts, err := s.accounts.ListCloudAccounts(ctx, config.CloudAccountFilter{Provider: &provider}) + enabled := true + accounts, err := s.accounts.ListCloudAccounts(ctx, config.CloudAccountFilter{Provider: &provider, Enabled: &enabled}) if err != nil { return nil, fmt.Errorf("list cloud accounts: %w", err) } for i := range accounts { - if accounts[i].Provider == "aws" { + if accounts[i].Provider == provider && accounts[i].Enabled { return &accounts[i], nil } } diff --git a/internal/commitmentopts/service_test.go b/internal/commitmentopts/service_test.go index c4ca48ed1..78e281f52 100644 --- a/internal/commitmentopts/service_test.go +++ b/internal/commitmentopts/service_test.go @@ -9,6 +9,8 @@ import ( "github.com/LeanerCloud/cloud-commitments-platform/internal/config" "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/opensearch" + opensearchtypes "github.com/aws/aws-sdk-go-v2/service/opensearch/types" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -388,3 +390,66 @@ func (p proberFunc) Service() string { return p.name } func (p proberFunc) Probe(ctx context.Context, _ aws.Config) ([]Combo, error) { return p.fn() } + +// A capped OpenSearch probe persists no opensearch combos, so a legitimate +// 3-year all-upfront save is allowed instead of rejected as unsold. +func TestService_Validate_CappedProbeStaysPermissive(t *testing.T) { + // Every page carries only the 1-year no-upfront offering and another page. + osFake := &fakeOpenSearch{fn: func(*opensearch.DescribeReservedInstanceOfferingsInput) (*opensearch.DescribeReservedInstanceOfferingsOutput, error) { + return &opensearch.DescribeReservedInstanceOfferingsOutput{ + ReservedInstanceOfferings: []opensearchtypes.ReservedInstanceOffering{{ + InstanceType: opensearchtypes.OpenSearchPartitionInstanceType(probeTargetOpenSearch), + Duration: 31536000, + PaymentOption: opensearchtypes.ReservedInstancePaymentOptionNoUpfront, + }}, + NextToken: aws.String("more"), + }, nil + }} + capped := &OpenSearchProber{NewClient: func(aws.Config) OpenSearchDescribeOfferings { return osFake }} + rds := &stubProber{name: "rds", combos: []Combo{{Provider: "aws", Service: "rds", TermYears: 1, Payment: "all-upfront"}}} + store := &fakeStore{} + svc := New(store, &fakeAccounts{accounts: []config.CloudAccount{awsAccount("a")}}, noopBuildConfig, []Prober{capped, rds}) + + ok, err := svc.Validate(context.Background(), "aws", "opensearch", 3, "all-upfront") + require.NoError(t, err) + assert.True(t, ok) +} + +// filterAccounts applies CloudAccountFilter the way the Postgres store does, +// and records the filter it was given. +type filterAccounts struct { + got config.CloudAccountFilter + accounts []config.CloudAccount +} + +func (f *filterAccounts) ListCloudAccounts(_ context.Context, filter config.CloudAccountFilter) ([]config.CloudAccount, error) { + f.got = filter + var out []config.CloudAccount + for _, a := range f.accounts { + if (filter.Provider == nil || a.Provider == *filter.Provider) && (filter.Enabled == nil || a.Enabled == *filter.Enabled) { + out = append(out, a) + } + } + return out, nil +} + +// A disabled AWS account listed first must not be the one probed (A05-010). +func TestService_FindAWSAccount_SkipsDisabled(t *testing.T) { + disabled := awsAccount("disabled") + disabled.Enabled = false + accounts := &filterAccounts{accounts: []config.CloudAccount{disabled, awsAccount("enabled")}} + svc := New(&fakeStore{}, accounts, noopBuildConfig, nil) + + got, err := svc.findAWSAccount(context.Background()) + require.NoError(t, err) + require.NotNil(t, got) + assert.Equal(t, "enabled", got.ID) + require.NotNil(t, accounts.got.Enabled, "the store query must ask for enabled accounts only") + assert.True(t, *accounts.got.Enabled) + + // A lister that ignores the filter still cannot hand back a disabled account. + svc = New(&fakeStore{}, &fakeAccounts{accounts: []config.CloudAccount{disabled}}, noopBuildConfig, nil) + got, err = svc.findAWSAccount(context.Background()) + require.NoError(t, err) + assert.Nil(t, got) +} diff --git a/internal/purchase/finalize_revocations.go b/internal/purchase/finalize_revocations.go index a91d80955..e89ee3400 100644 --- a/internal/purchase/finalize_revocations.go +++ b/internal/purchase/finalize_revocations.go @@ -2,6 +2,7 @@ package purchase import ( "context" + "fmt" "time" "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" @@ -42,6 +43,11 @@ var finalizeRevocationBackoffs = []time.Duration{ // are logged and counted in FinalizeResult.Errored but do not abort the // sweep — the sweep continues to the next row so a single stuck row does not // block finalization of all in-flight rows. +// +// Cancellation: the backoff between retries honors ctx. When ctx is done the +// row being retried is counted as Errored, the counts so far are logged, and +// the sweep returns them with an error wrapping ctx.Err(), so a Lambda near +// its deadline stops instead of sleeping past it (issue #243). func (m *Manager) FinalizeInFlightRevocations(ctx context.Context) (*FinalizeResult, error) { rows, err := m.config.GetPurchaseHistoryInFlight(ctx) if err != nil { @@ -52,15 +58,12 @@ func (m *Manager) FinalizeInFlightRevocations(ctx context.Context) (*FinalizeRes now := time.Now().UTC() for _, record := range rows { - markErr := m.config.MarkPurchaseRevoked(ctx, record.PurchaseID, now, "direct-api", "", nil, "") - for attempt, backoff := range finalizeRevocationBackoffs { - if markErr == nil { - break - } - logging.Warnf("finalize_revocations: MarkPurchaseRevoked attempt %d for %s failed: %v (retrying in %s)", - attempt+1, record.PurchaseID, markErr, backoff) - time.Sleep(backoff) - markErr = m.config.MarkPurchaseRevoked(ctx, record.PurchaseID, now, "direct-api", "", nil, "") + markErr, ctxErr := m.markRevokedWithRetry(ctx, record.PurchaseID, now) + if ctxErr != nil { + result.Errored++ + logging.Errorf("finalize_revocations: sweep stopped at purchase_id=%s: %v (found=%d finalized=%d errored=%d)", + record.PurchaseID, ctxErr, result.Found, result.Finalized, result.Errored) + return result, fmt.Errorf("finalize_revocations: sweep canceled: %w", ctxErr) } if markErr != nil { logging.Errorf("finalize_revocations: MarkPurchaseRevoked for %s failed after %d attempts: %v", @@ -74,3 +77,26 @@ func (m *Manager) FinalizeInFlightRevocations(ctx context.Context) (*FinalizeRes return result, nil } + +// markRevokedWithRetry calls MarkPurchaseRevoked, retrying after each +// finalizeRevocationBackoffs delay. It returns the last MarkPurchaseRevoked +// error, or a non-nil ctxErr when ctx was done during a backoff. +func (m *Manager) markRevokedWithRetry(ctx context.Context, purchaseID string, now time.Time) (markErr, ctxErr error) { + markErr = m.config.MarkPurchaseRevoked(ctx, purchaseID, now, "direct-api", "", nil, "") + for attempt, backoff := range finalizeRevocationBackoffs { + if markErr == nil { + return nil, nil + } + logging.Warnf("finalize_revocations: MarkPurchaseRevoked attempt %d for %s failed: %v (retrying in %s)", + attempt+1, purchaseID, markErr, backoff) + timer := time.NewTimer(backoff) + select { + case <-ctx.Done(): + timer.Stop() + return markErr, ctx.Err() + case <-timer.C: + } + markErr = m.config.MarkPurchaseRevoked(ctx, purchaseID, now, "direct-api", "", nil, "") + } + return markErr, nil +} diff --git a/internal/purchase/finalize_revocations_test.go b/internal/purchase/finalize_revocations_test.go new file mode 100644 index 000000000..aa0d9b97f --- /dev/null +++ b/internal/purchase/finalize_revocations_test.go @@ -0,0 +1,103 @@ +package purchase + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// shortFinalizeBackoffs swaps the retry delays for the test's duration. +func shortFinalizeBackoffs(t *testing.T, d ...time.Duration) { + t.Helper() + orig := finalizeRevocationBackoffs + finalizeRevocationBackoffs = d + t.Cleanup(func() { finalizeRevocationBackoffs = orig }) +} + +func inFlightRows(ids ...string) []*config.PurchaseHistoryRecord { + rows := make([]*config.PurchaseHistoryRecord, 0, len(ids)) + for _, id := range ids { + rows = append(rows, &config.PurchaseHistoryRecord{PurchaseID: id}) + } + return rows +} + +func expectMark(store *MockConfigStore, id string, err error) *mock.Call { + return store.On("MarkPurchaseRevoked", mock.Anything, id, mock.Anything, "direct-api", "", (*float64)(nil), ""). + Return(err).Once() +} + +// One row finalizes first try, one on a retry, one never: the counts must say +// exactly that, and every row must be stamped with the same sweep timestamp. +func TestFinalizeInFlightRevocations_Accounting(t *testing.T) { + shortFinalizeBackoffs(t, time.Millisecond, time.Millisecond) + ctx := context.Background() + store := new(MockConfigStore) + store.On("GetPurchaseHistoryInFlight", ctx).Return(inFlightRows("first", "retry", "never"), nil) + dbDown := errors.New("db down") + expectMark(store, "first", nil) + expectMark(store, "retry", dbDown) + expectMark(store, "retry", nil) + expectMark(store, "never", dbDown) + expectMark(store, "never", dbDown) + expectMark(store, "never", dbDown) + m := &Manager{config: store} + + res, err := m.FinalizeInFlightRevocations(ctx) + require.NoError(t, err) + assert.Equal(t, FinalizeResult{Found: 3, Finalized: 2, Errored: 1}, *res) + store.AssertExpectations(t) + store.AssertNumberOfCalls(t, "MarkPurchaseRevoked", 6) + + var stamps []time.Time + for _, c := range store.Calls { + if c.Method == "MarkPurchaseRevoked" { + stamps = append(stamps, c.Arguments.Get(2).(time.Time)) + } + } + for _, ts := range stamps { + assert.True(t, ts.Equal(stamps[0]), "every row in one sweep shares one revokedAt") + } +} + +func TestFinalizeInFlightRevocations_ListErrorIsReturned(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + store.On("GetPurchaseHistoryInFlight", ctx).Return(nil, errors.New("list failed")) + m := &Manager{config: store} + + res, err := m.FinalizeInFlightRevocations(ctx) + require.EqualError(t, err, "list failed") + assert.Nil(t, res) + store.AssertNotCalled(t, "MarkPurchaseRevoked", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything) +} + +// A context canceled during a backoff stops the sweep promptly: the row being +// retried counts as Errored, later rows are not attempted, and the partial +// counts come back with an error wrapping context.Canceled. +func TestFinalizeInFlightRevocations_CanceledDuringBackoffStopsSweep(t *testing.T) { + shortFinalizeBackoffs(t, 3*time.Second, 3*time.Second) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + store := new(MockConfigStore) + store.On("GetPurchaseHistoryInFlight", ctx).Return(inFlightRows("ok", "stuck", "later"), nil) + expectMark(store, "ok", nil) + expectMark(store, "stuck", errors.New("db down")).Run(func(mock.Arguments) { cancel() }) + m := &Manager{config: store} + + start := time.Now() + res, err := m.FinalizeInFlightRevocations(ctx) + elapsed := time.Since(start) + + require.ErrorIs(t, err, context.Canceled) + require.NotNil(t, res) + assert.Equal(t, FinalizeResult{Found: 3, Finalized: 1, Errored: 1}, *res) + assert.Less(t, elapsed, time.Second, "the sweep must not sleep through the backoff after cancellation") + store.AssertNumberOfCalls(t, "MarkPurchaseRevoked", 2) +} diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index adba19c9e..ed9b9f55e 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -485,10 +485,14 @@ func armedRedriveRefusal(exec *config.PurchaseExecution) error { // purchase APIs take no client token, so a lost response may have bought the // commitment and a second attempt would buy another (MON-02). The provider // sends the call once and reports "purchase outcome unknown". -// - Azure savings-plans: the OrderAlias API uses time.Now().UnixNano() as -// the alias name; there is no server-side idempotency key and no -// tag-based lookup implemented yet. Re-driving would create a duplicate -// savings plan. +// - Azure savings-plans: the pinned client names the order alias after the +// idempotency token when one is set (a timestamp name only on the +// empty-token CLI path), but no test or provider doc here confirms that a +// repeat PUT of the same alias collapses onto the first order, and there +// is no tag-based lookup. Until that is verified, a re-drive is treated as +// able to buy a duplicate savings plan. The refusal covers exactly the +// slugs that dispatch to the Azure savings-plans client (pinned by +// TestRedriveGuardReachMatchesDispatchReach). // - Any provider this function does not recognize, rather than assuming a // guard exists. // @@ -529,9 +533,10 @@ func recRedriveRefusalReason(rec config.RecommendationRecord) string { } return "" case "azure": - // Azure savings-plans uses a timestamp-based alias name and has no - // server-side idempotency key, so a re-drive would create a duplicate. - // All other Azure services use DoIdempotentPurchaseTwoStep (#729). + // Azure savings-plans: a repeat order-alias PUT is not confirmed to + // collapse onto the first order (see RedriveRefusalReason), so a + // re-drive may create a duplicate. All other Azure services use + // DoIdempotentPurchaseTwoStep (#729). if rec.Service == "savingsplans" || rec.Service == "savings-plans" { return "Azure savings plans have no provider-side duplicate guard, so re-driving this purchase would buy a second savings plan that cannot be canceled" } @@ -710,9 +715,10 @@ func (m *Manager) safeFail(ctx context.Context, exec *config.PurchaseExecution) // Retry step required by the old safe-fail path. See allRecsSafeToRedrive for // which provider/service combinations are eligible. // -// Safe-fail path: Azure savings-plans recs are excluded because the OrderAlias -// API uses a timestamp-based alias name with no idempotency key. Executions -// without a stable ExecutionID (legacy rows) also fall through because +// Safe-fail path: Azure savings-plans recs are excluded because a repeat +// order-alias PUT is not confirmed to be deduplicated (see +// RedriveRefusalReason). Executions without a stable ExecutionID (legacy +// rows) also fall through because // idempotencyLineageKey(exec) falls back to "" for them and // DeriveIdempotencyToken("", i) would produce the same token set for every // such row. These fall through to the original behavior: the row is atomically @@ -772,8 +778,9 @@ func (m *Manager) RecoverStrandedApprovals(ctx context.Context) (int, error) { // wait for the token-link approval path; the scheduler and SQS paths must // never bypass that gate. // - All other pending/notified rows require the owning plan to have -// AutoPurchase=true. A plan-fetch error is propagated so the caller can -// fail closed rather than defaulting to "execute". +// AutoPurchase=true. A plan-fetch error, or a missing plan, is propagated +// so the caller can fail closed rather than defaulting to "execute". A +// direct-execute row (no PlanID) is never eligible and is not an error. // - The global kill switch (GlobalConfig.AutoPurchaseEnabled) must be on. A // read error is propagated. Only AutoPurchase plans reach this check, so // manual purchases are unaffected. @@ -793,11 +800,11 @@ func (m *Manager) executableByScheduler(ctx context.Context, exec *config.Purcha if exec.Source == common.PurchaseSourceWeb { return false, nil } - plan, err := m.config.GetPurchasePlan(ctx, exec.PlanID) + plan, err := m.autoPurchasePlan(ctx, exec) if err != nil { - return false, fmt.Errorf("failed to fetch plan %s for AutoPurchase gate: %w", exec.PlanID, err) + return false, err } - if !plan.AutoPurchase { + if plan == nil { return false, nil } enabled, err := m.autoPurchaseEnabled(ctx) @@ -819,6 +826,33 @@ func (m *Manager) executableByScheduler(ctx context.Context, exec *config.Purcha return true, nil } +// autoPurchasePlan returns the plan that owns exec when that plan has +// AutoPurchase=true, nil when the row cannot be auto-purchased, and an error +// when the plan cannot be read (the caller fails closed). +// +// A direct-execute row (PlanID == "") has no plan to arm it, so it is never +// auto-purchased. Short-circuit it before the store: the Postgres UUID column +// rejects "" with SQLSTATE 22P02, which every scheduler tick would otherwise +// count as a failure for as long as the row stays pending (issue #243). A nil +// plan with no error is refused as not found, as executePurchase does, rather +// than dereferenced. +func (m *Manager) autoPurchasePlan(ctx context.Context, exec *config.PurchaseExecution) (*config.PurchasePlan, error) { + if exec.PlanID == "" { + return nil, nil + } + plan, err := m.config.GetPurchasePlan(ctx, exec.PlanID) + if err != nil { + return nil, fmt.Errorf("failed to fetch plan %s for AutoPurchase gate: %w", exec.PlanID, err) + } + if plan == nil { + return nil, fmt.Errorf("plan %s not found for AutoPurchase gate", exec.PlanID) + } + if !plan.AutoPurchase { + return nil, nil + } + return plan, nil +} + // autoPurchaseEnabled reads the global auto-purchase kill switch. It fails // closed: a read error is returned, never read as "enabled". func (m *Manager) autoPurchaseEnabled(ctx context.Context) (bool, error) { diff --git a/internal/purchase/money_path_regression_test.go b/internal/purchase/money_path_regression_test.go index 5233d8e0d..1fbac842b 100644 --- a/internal/purchase/money_path_regression_test.go +++ b/internal/purchase/money_path_regression_test.go @@ -303,6 +303,7 @@ func TestSQSRedeliveryDoesNotDoubleExecute(t *testing.T) { newPending := func() *config.PurchaseExecution { return &config.PurchaseExecution{ ExecutionID: "exec-dup", + PlanID: "plan-dup", // a plan-less row is never auto-executed IdempotencyKey: "lineage-dup", Status: "notified", Recommendations: []config.RecommendationRecord{ diff --git a/internal/purchase/reaper.go b/internal/purchase/reaper.go index 883f3210c..e7ba2a931 100644 --- a/internal/purchase/reaper.go +++ b/internal/purchase/reaper.go @@ -271,9 +271,9 @@ func (m *Manager) reapOne(ctx context.Context, exec *config.PurchaseExecution, r // the reaper attribution. Bump Errored so ops can track it. // // "safe to retry" is only appended when every recommendation in the - // execution uses an idempotent provider API. Azure savings-plans use a - // timestamp-based alias name with no server-side idempotency key, so - // retrying a reaped Azure SP row risks creating a duplicate commitment. + // execution uses an idempotent provider API. Azure savings-plans are not + // confirmed to deduplicate a repeat order-alias PUT (RedriveRefusalReason), + // so retrying a reaped Azure SP row risks creating a duplicate commitment. safeMsg := "" if allRecsSafeToRedrive(exec) { safeMsg = "; safe to retry" diff --git a/internal/purchase/step_cap_test.go b/internal/purchase/step_cap_test.go index 0f8e58a51..ca35a40ba 100644 --- a/internal/purchase/step_cap_test.go +++ b/internal/purchase/step_cap_test.go @@ -111,6 +111,8 @@ func TestScheduler_RefusesBeforeClaim(t *testing.T) { {"NULL cap", capped(nil), dueRecExec("notified", upfrontRec(10)), autoPurchaseOnCfg(), nil, 1}, {"over cap", capped(f64(5)), dueRecExec("notified", upfrontRec(10)), autoPurchaseOnCfg(), nil, 1}, {"zero commitment", capped(f64(1000)), dueRecExec("notified", upfrontRec(0)), autoPurchaseOnCfg(), nil, 1}, + // The store interface permits (nil, nil); the gate must refuse, not panic (A05-013). + {"plan missing", nil, dueRecExec("notified", upfrontRec(10)), autoPurchaseOnCfg(), nil, 1}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -140,6 +142,7 @@ func TestSQSExecutePath_RefusesBeforeClaim(t *testing.T) { {"NULL cap", capped(nil), dueRecExec("notified", upfrontRec(10)), autoPurchaseOnCfg(), nil}, {"over cap", capped(f64(5)), dueRecExec("notified", upfrontRec(10)), autoPurchaseOnCfg(), nil}, {"zero commitment", capped(f64(1000)), dueRecExec("notified", upfrontRec(0)), autoPurchaseOnCfg(), nil}, + {"plan missing", nil, dueRecExec("notified", upfrontRec(10)), autoPurchaseOnCfg(), nil}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -230,3 +233,24 @@ func TestSendUpcomingPurchaseNotifications_KillSwitch(t *testing.T) { require.Error(t, err) }) } + +// A direct-execute row (no PlanID) left pending is never auto-purchased, and +// the gate must not hand "" to the store: Postgres rejects it with SQLSTATE +// 22P02, which every tick would count as a failure (issue #243). +func TestScheduler_DirectExecuteRowSkippedWithoutPlanLookup(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + exec := dueRecExec("pending", upfrontRec(10)) + exec.PlanID = "" + store.On("GetStaleApprovedExecutions", ctx, mock.Anything).Return([]config.PurchaseExecution{}, nil) + store.On("GetPendingExecutions", ctx).Return([]config.PurchaseExecution{exec}, nil) + store.On("GetPurchasePlan", ctx, "").Return(nil, errors.New(`ERROR: invalid input syntax for type uuid: "" (SQLSTATE 22P02)`)) + manager := &Manager{config: store, email: new(MockEmailSender)} + + result, err := manager.ProcessScheduledPurchases(ctx) + require.NoError(t, err) + assert.Equal(t, 0, result.Failed, "a plan-less row must not poison the tick's failure count: %v", result.Errors) + assert.Equal(t, 0, result.Processed) + store.AssertNotCalled(t, "GetPurchasePlan", mock.Anything, mock.Anything) + store.AssertNotCalled(t, "TransitionExecutionStatus", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything) +}