From 89c1363d3ffef593701f37a88dd19481146a3c24 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Mon, 19 Oct 2026 12:54:36 +0100 Subject: [PATCH] fix(exchange): alarm on stale processing RI exchanges and let an operator resolve them A run that crashes between the provider call and its ledger write leaves an ri_exchange_history row in processing. Since #690 a processing row counts against the daily cap on every later day, and nothing reconciled it, so a crashed run could block every later exchange indefinitely. - The scheduled ri_exchange_reshape run now logs an ERROR (the application error alarm channel) for every processing row unchanged for over an hour, naming the row, its held amount and how to resolve it. It runs before the enabled check and never settles or releases anything itself: whether the exchange happened at AWS is unknown, so the hold stays counted. - POST /api/ri-exchange/resolve/{id} (execute:ri-exchange, same account scope and region constraints as direct execution, signed-in user only) records an operator's decision after checking AWS: "completed" keeps the reserved ceiling counted on today's UTC day; "failed" releases the hold and needs acknowledge_unknown_outcome. - PostgresStore.ResolveStaleRIExchange locks the row and refuses unless it is still processing and unchanged for 6h on the database clock, the cloud check names the row's own account and region and postdates its last change, and (for a release) no provider exchange id is recorded. A row a late writer settled is never overwritten; identical retries replay. Closes #692 Co-Authored-By: Claude Opus 5.5 --- internal/api/handler_ri_exchange_resolve.go | 218 +++++++++++++ .../api/handler_ri_exchange_resolve_test.go | 224 +++++++++++++ internal/api/router.go | 7 + .../store_postgres_ri_exchange_resolve.go | 219 +++++++++++++ ...es_ri_exchange_resolve_integration_test.go | 299 ++++++++++++++++++ internal/server/handler_ri_exchange.go | 34 ++ .../server/handler_ri_exchange_stale_test.go | 126 ++++++++ 7 files changed, 1127 insertions(+) create mode 100644 internal/api/handler_ri_exchange_resolve.go create mode 100644 internal/api/handler_ri_exchange_resolve_test.go create mode 100644 internal/config/store_postgres_ri_exchange_resolve.go create mode 100644 internal/config/store_postgres_ri_exchange_resolve_integration_test.go create mode 100644 internal/server/handler_ri_exchange_stale_test.go diff --git a/internal/api/handler_ri_exchange_resolve.go b/internal/api/handler_ri_exchange_resolve.go new file mode 100644 index 000000000..7143563b9 --- /dev/null +++ b/internal/api/handler_ri_exchange_resolve.go @@ -0,0 +1,218 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "regexp" + "strings" + "time" + "unicode/utf8" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/aws/aws-lambda-go/events" +) + +// Operator resolve of a stale processing RI exchange (#692). A run that +// crashed between the provider call and its ledger write leaves the row +// processing, and a processing row holds its reserved ceiling against the +// daily cap on every later day (#657). Only a person who checked AWS can say +// whether the exchange happened, so this endpoint records their decision; the +// store refuses rows that a running writer may still settle. + +const ( + riExchangeResolveReasonMin = 10 + riExchangeResolveReasonMax = 300 + riExchangeResolveBodyMax = 4096 +) + +var ( + riExchangeResolveAccountRe = regexp.MustCompile(`^\d{12}$`) + riExchangeResolveRegionRe = regexp.MustCompile(`^[a-z0-9-]{1,32}$`) + riExchangeResolveExchangeRe = regexp.MustCompile(`^[A-Za-z0-9-]{1,100}$`) +) + +// riExchangeResolveRequest is the body of POST /api/ri-exchange/resolve/{id}. +// outcome "failed" releases the hold and needs acknowledge_unknown_outcome; +// "completed" keeps the reserved amount on today's books and may name the +// exchange id the operator found. +type riExchangeResolveRequest struct { + Outcome string `json:"outcome"` + Reason string `json:"reason"` + ExchangeID string `json:"exchange_id"` + Acknowledge bool `json:"acknowledge_unknown_outcome"` + CloudCheck *riExchangeResolveCloudCheck `json:"cloud_check"` +} + +type riExchangeResolveCloudCheck struct { + AccountID string `json:"checked_account_id"` + Region string `json:"checked_region"` + CheckedAt string `json:"checked_at"` +} + +type riExchangeResolveResponse struct { + ID string `json:"id"` + Status string `json:"status"` + PaymentDue string `json:"payment_due"` + Note string `json:"note"` + Replayed bool `json:"replayed"` +} + +func (h *Handler) resolveRIExchange(ctx context.Context, req *events.LambdaFunctionURLRequest, id string) (any, error) { + session, err := h.requirePermission(ctx, req, "execute", "ri-exchange") + if err != nil { + return nil, err + } + // The decision is stamped on the row as transitioned_by, so it needs an + // identified person, not the stateless admin key. + if session.UserID == apiKeyAdminUserID { + return nil, NewClientError(403, "resolving an RI exchange needs a signed-in user or a user API key") + } + if idErr := validateUUID(id); idErr != nil { + return nil, idErr + } + // Same deployment-account scope as approvals: out of scope reads as missing. + if scopeErr := h.requireReshapeAccountVisible(ctx, session); scopeErr != nil { + return nil, scopeErr + } + body, check, err := decodeRIExchangeResolve(req.Body) + if err != nil { + return nil, err + } + if constraintErr := h.requireRIExchangeResolveConstraints(ctx, session, check.Region); constraintErr != nil { + return nil, constraintErr + } + store, ok := h.config.(config.RIExchangeResolveStore) + if !ok { + return nil, NewClientError(503, "RI exchange resolve storage is unavailable") + } + res, err := store.ResolveStaleRIExchange(ctx, config.RIExchangeResolve{ + ID: id, Outcome: body.Outcome, ExchangeID: body.ExchangeID, Note: body.note(check), + Check: check, Actor: resolveCreatorUserID(session), StaleAfter: config.RIExchangeResolveStaleAfter, + }) + if mapped := mapRIExchangeResolveError(err); mapped != nil { + return nil, mapped + } + if !res.Replayed { + logging.Warnf("RI exchange resolved by operator: id=%s actor=%q via=%s processing -> %s payment_due=%s note=%q", + id, session.UserID, credentialKind(ctx, session), res.Status, res.PaymentDue, res.Note) + } + return riExchangeResolveResponse{ID: res.ID, Status: res.Status, PaymentDue: res.PaymentDue, Note: res.Note, Replayed: res.Replayed}, nil +} + +// requireRIExchangeResolveConstraints applies the Constraints on the granting +// execute:ri-exchange permission, as the direct execute endpoint does. The +// region is the operator's checked_region, which the store requires to equal +// the exchange's own region. +func (h *Handler) requireRIExchangeResolveConstraints(ctx context.Context, session *Session, region string) error { + cloudAccountID, err := h.resolveReshapeCloudAccountIDOrUnattributed(ctx) + if err != nil { + return fmt.Errorf("failed to resolve cloud account scope: %w", err) + } + return h.requirePermissionConstraints(ctx, session, "execute", "ri-exchange", []auth.PermissionConstraints{{ + AccountIDs: []string{cloudAccountID}, + Providers: []string{string(common.ProviderAWS)}, + Services: []string{string(common.ServiceEC2)}, + Regions: []string{region}, + }}) +} + +func mapRIExchangeResolveError(err error) error { + switch { + case err == nil: + return nil + case errors.Is(err, config.ErrRIExchangeResolveNotFound): + return errNotFound + case errors.Is(err, config.ErrRIExchangeResolveInvalid): + return NewClientError(400, err.Error()) + case errors.Is(err, config.ErrRIExchangeResolveConflict): + return NewClientError(409, err.Error()) + } + return err +} + +func decodeRIExchangeResolve(raw string) (riExchangeResolveRequest, config.RIExchangeCloudCheck, error) { + var body riExchangeResolveRequest + if len(raw) > riExchangeResolveBodyMax { + return body, config.RIExchangeCloudCheck{}, NewClientError(400, "RI exchange resolve body is too large") + } + decoder := json.NewDecoder(strings.NewReader(raw)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&body); err != nil { + return body, config.RIExchangeCloudCheck{}, NewClientError(400, "Invalid RI exchange resolve body") + } + if err := decoder.Decode(new(any)); err != io.EOF { + return body, config.RIExchangeCloudCheck{}, NewClientError(400, "Expected one RI exchange resolve body") + } + body.Reason = strings.TrimSpace(body.Reason) + if err := body.validate(); err != nil { + return body, config.RIExchangeCloudCheck{}, err + } + check, err := body.cloudCheck() + return body, check, err +} + +// validate checks the trimmed body's shape; the store decides the rest. +func (b riExchangeResolveRequest) validate() error { + switch { + case b.Outcome != config.StatusFailed && b.Outcome != config.StatusCompleted: + return NewClientError(400, `outcome must be "failed" or "completed"`) + case utf8.RuneCountInString(b.Reason) < riExchangeResolveReasonMin || utf8.RuneCountInString(b.Reason) > riExchangeResolveReasonMax: + return NewClientError(400, fmt.Sprintf("reason must be %d to %d characters", riExchangeResolveReasonMin, riExchangeResolveReasonMax)) + case hasControlRune(b.Reason): + return NewClientError(400, "reason must not contain control characters") + case b.CloudCheck == nil: + return NewClientError(400, "cloud_check is required: say where you verified the exchange at AWS") + } + return b.validateOutcomeFields() +} + +func (b riExchangeResolveRequest) validateOutcomeFields() error { + if b.Outcome == config.StatusCompleted { + if b.ExchangeID != "" && !riExchangeResolveExchangeRe.MatchString(b.ExchangeID) { + return NewClientError(400, "exchange_id must be a provider exchange id") + } + return nil + } + if b.ExchangeID != "" { + return NewClientError(400, "exchange_id means the exchange happened; resolve it as completed") + } + if !b.Acknowledge { + return NewClientError(400, "acknowledge_unknown_outcome is required to release a hold: CUDly cannot see whether the exchange happened") + } + return nil +} + +// cloudCheck validates the shape of the operator's AWS-side check. Whether it +// names the exchange's own account and region, and postdates its last change, +// is decided by the store under the row lock. +func (b riExchangeResolveRequest) cloudCheck() (config.RIExchangeCloudCheck, error) { + c := b.CloudCheck + at, err := time.Parse(time.RFC3339, c.CheckedAt) + if err != nil { + return config.RIExchangeCloudCheck{}, NewClientError(400, "checked_at must be an RFC 3339 timestamp") + } + if !riExchangeResolveAccountRe.MatchString(c.AccountID) || !riExchangeResolveRegionRe.MatchString(c.Region) { + return config.RIExchangeCloudCheck{}, NewClientError(400, "checked_account_id must be a 12-digit AWS account id and checked_region an AWS region") + } + return config.RIExchangeCloudCheck{AccountID: c.AccountID, Region: c.Region, CheckedAt: at.UTC()}, nil +} + +// note is the exact text stored in ri_exchange_history.error. The fields are in +// a fixed order and quoted, so a repeat of the same request produces the same +// string and a different one cannot. +func (b riExchangeResolveRequest) note(check config.RIExchangeCloudCheck) string { + s := fmt.Sprintf("operator:%s: account=%q region=%q checked_at=%s", b.Outcome, check.AccountID, check.Region, check.CheckedAt.Format(time.RFC3339)) + if b.Outcome == config.StatusFailed { + s += " ack=true" + } + if b.ExchangeID != "" { + s += fmt.Sprintf(" exchange_id=%q", b.ExchangeID) + } + return s + fmt.Sprintf(" reason=%q", b.Reason) +} diff --git a/internal/api/handler_ri_exchange_resolve_test.go b/internal/api/handler_ri_exchange_resolve_test.go new file mode 100644 index 000000000..5610ce54b --- /dev/null +++ b/internal/api/handler_ri_exchange_resolve_test.go @@ -0,0 +1,224 @@ +package api + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/aws/aws-lambda-go/events" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// Handler tests for POST /api/ri-exchange/resolve/{id} (#692). The store is a +// FIXTURE; the row lock, staleness and cloud-check rules are covered against +// real PostgreSQL in internal/config. + +const resolveRIID = "6f1c2a8e-0d4b-4c1e-9a3f-692000000001" + +var ( + riExecute = []auth.Permission{{Action: auth.ActionExecute, Resource: "ri-exchange"}} + riResolveAt = time.Date(2026, 10, 19, 8, 0, 0, 0, time.UTC) +) + +type resolveRIStore struct { + *MockConfigStore + called int + got config.RIExchangeResolve + res *config.RIExchangeResolveResult + err error +} + +func (s *resolveRIStore) ResolveStaleRIExchange(_ context.Context, r config.RIExchangeResolve) (*config.RIExchangeResolveResult, error) { + s.called++ + s.got = r + if s.res == nil && s.err == nil { + return &config.RIExchangeResolveResult{ID: r.ID, Status: r.Outcome, PaymentDue: "1000.000000", Note: r.Note}, nil + } + return s.res, s.err +} + +// riResolveAccount renders the 12-digit account id without a literal the +// git-secrets AWS account-id pattern would flag. +func riResolveAccount() string { return fmt.Sprintf("%012d", 692000000001) } + +func riResolveBody(outcome, extra string) string { + return fmt.Sprintf(`{"outcome":%q,"reason":"source RIs still active in the EC2 console"%s,`+ + `"cloud_check":{"checked_account_id":%q,"checked_region":"us-east-1","checked_at":%q}}`, + outcome, extra, riResolveAccount(), riResolveAt.Format(time.RFC3339)) +} + +var riResolveFailedBody = riResolveBody("failed", `,"acknowledge_unknown_outcome":true`) + +func riResolveCall(t *testing.T, caller apgCaller, store *resolveRIStore, body string) (any, error) { + t.Helper() + h, req := caller.handlerAndReq(store.MockConfigStore, body) + h.config = store + h.reshapeAccountResolver = func(context.Context) (string, error) { return scopedInAccount, nil } + return h.resolveRIExchange(context.Background(), req, resolveRIID) +} + +func TestResolveRIExchange_PassesOperatorDecisionToStore(t *testing.T) { + store := &resolveRIStore{MockConfigStore: new(MockConfigStore)} + out, err := riResolveCall(t, apgCaller{perms: riExecute}, store, riResolveFailedBody) + require.NoError(t, err) + require.Equal(t, 1, store.called) + got := store.got + assert.Equal(t, resolveRIID, got.ID) + assert.Equal(t, config.StatusFailed, got.Outcome) + assert.Empty(t, got.ExchangeID) + assert.Equal(t, config.RIExchangeResolveStaleAfter, got.StaleAfter) + assert.Equal(t, config.RIExchangeCloudCheck{AccountID: riResolveAccount(), Region: "us-east-1", CheckedAt: riResolveAt}, got.Check) + require.NotNil(t, got.Actor) + assert.Equal(t, apgUserID, *got.Actor, "the decision is attributed to the signed-in operator") + assert.Equal(t, fmt.Sprintf(`operator:failed: account=%q region="us-east-1" checked_at=2026-10-19T08:00:00Z ack=true reason="source RIs still active in the EC2 console"`, riResolveAccount()), got.Note) + resp, ok := out.(riExchangeResolveResponse) + require.True(t, ok) + assert.Equal(t, config.StatusFailed, resp.Status) + + completed := &resolveRIStore{MockConfigStore: new(MockConfigStore)} + _, err = riResolveCall(t, apgCaller{perms: riExecute}, completed, riResolveBody("completed", `,"exchange_id":"riex-0a1b2c"`)) + require.NoError(t, err) + assert.Equal(t, config.StatusCompleted, completed.got.Outcome) + assert.Equal(t, "riex-0a1b2c", completed.got.ExchangeID) + assert.Contains(t, completed.got.Note, `exchange_id="riex-0a1b2c"`) + assert.NotContains(t, completed.got.Note, "ack=true") +} + +func TestResolveRIExchange_Authorization(t *testing.T) { + regionScoped := func(region string) []auth.Permission { + return []auth.Permission{{Action: auth.ActionExecute, Resource: "ri-exchange", + Constraints: &auth.PermissionConstraints{Regions: []string{region}}}} + } + tests := []struct { + name string + caller apgCaller + want int // 0 = allowed + }{ + {"execute:ri-exchange constrained to the checked region", apgCaller{perms: regionScoped("us-east-1")}, 0}, + {"view-only user", apgCaller{perms: []auth.Permission{{Action: auth.ActionView, Resource: auth.ResourcePurchases}}}, 403}, + {"admin API key has no identity to stamp", apgCaller{key: true}, 403}, + {"execute:ri-exchange constrained to another region", apgCaller{perms: regionScoped("eu-west-1")}, 403}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + store := &resolveRIStore{MockConfigStore: new(MockConfigStore)} + _, err := riResolveCall(t, tc.caller, store, riResolveFailedBody) + if tc.want == 0 { + require.NoError(t, err) + assert.Equal(t, 1, store.called) + return + } + assert.Equal(t, tc.want, clientCode(t, err)) + assert.Zero(t, store.called, "a refused caller must not reach the store") + }) + } +} + +// A session whose allowed accounts exclude the deployment's account sees the +// enumeration-safe not-found, like the approval path. +func TestResolveRIExchange_OutOfScopeAccount(t *testing.T) { + store := &resolveRIStore{MockConfigStore: new(MockConfigStore)} + store.GetCloudAccountFn = func(_ context.Context, id string) (*config.CloudAccount, error) { + return &config.CloudAccount{ID: id, Provider: "aws"}, nil + } + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", mock.Anything, "tok").Return(&Session{UserID: apgUserID}, nil).Maybe() + mockAuth.grantPermissionsScoped(riExecute, []string{scopedInAccount}) + h := &Handler{config: store, auth: mockAuth, + reshapeAccountResolver: func(context.Context) (string, error) { return scopedOutAccount, nil }} + req := &events.LambdaFunctionURLRequest{Headers: map[string]string{"Authorization": "Bearer tok"}, Body: riResolveFailedBody} + _, err := h.resolveRIExchange(context.Background(), req, resolveRIID) + require.ErrorIs(t, err, errNotFound) + assert.Zero(t, store.called) +} + +func TestResolveRIExchange_BodyValidation(t *testing.T) { + check := func(acct, region, at string) string { + return fmt.Sprintf(`"cloud_check":{"checked_account_id":%q,"checked_region":%q,"checked_at":%q}`, acct, region, at) + } + good := check(riResolveAccount(), "us-east-1", riResolveAt.Format(time.RFC3339)) + reason := `"reason":"source RIs still active in the EC2 console"` + for name, body := range map[string]string{ + "empty": ``, + "unknown field": `{"outcome":"failed",` + reason + `,"acknowledge_unknown_outcome":true,` + good + `,"actor":"x"}`, + "two documents": riResolveFailedBody + riResolveFailedBody, + "bad outcome": `{"outcome":"canceled",` + reason + `,` + good + `}`, + "short reason": `{"outcome":"completed","reason":"ok",` + good + `}`, + "whitespace reason": `{"outcome":"completed","reason":" ",` + good + `}`, + "control char in reason": `{"outcome":"completed","reason":"source RIs\nstill active",` + good + `}`, + "missing cloud check": `{"outcome":"completed",` + reason + `}`, + "failed without ack": `{"outcome":"failed",` + reason + `,` + good + `}`, + "failed with exchange id": `{"outcome":"failed",` + reason + `,"acknowledge_unknown_outcome":true,"exchange_id":"riex-1",` + good + `}`, + "bad exchange id": `{"outcome":"completed",` + reason + `,"exchange_id":"riex 1",` + good + `}`, + "bad checked_at": `{"outcome":"completed",` + reason + `,` + check(riResolveAccount(), "us-east-1", "yesterday") + `}`, + "non-numeric account": `{"outcome":"completed",` + reason + `,` + check("acct-1", "us-east-1", "2026-10-19T08:00:00Z") + `}`, + "upper-case region": `{"outcome":"completed",` + reason + `,` + check(riResolveAccount(), "US East", "2026-10-19T08:00:00Z") + `}`, + "oversized body": `{"outcome":"completed","reason":"` + strings.Repeat("x", 5000) + `"}`, + "long reason": `{"outcome":"completed","reason":"` + strings.Repeat("x", 301) + `",` + good + `}`, + "cloud check null for fail": `{"outcome":"failed",` + reason + `,"acknowledge_unknown_outcome":true,"cloud_check":null}`, + } { + t.Run(name, func(t *testing.T) { + store := &resolveRIStore{MockConfigStore: new(MockConfigStore)} + _, err := riResolveCall(t, apgCaller{perms: riExecute}, store, body) + assert.Equal(t, 400, clientCode(t, err)) + assert.Zero(t, store.called, "an invalid body must not reach the store") + }) + } +} + +func TestResolveRIExchange_BadIDIsRefusedBeforeTheStore(t *testing.T) { + store := &resolveRIStore{MockConfigStore: new(MockConfigStore)} + h, req := apgCaller{perms: riExecute}.handlerAndReq(store.MockConfigStore, riResolveFailedBody) + h.config = store + h.reshapeAccountResolver = func(context.Context) (string, error) { return scopedInAccount, nil } + _, err := h.resolveRIExchange(context.Background(), req, "not-a-uuid") + assert.Equal(t, 400, clientCode(t, err)) + assert.Zero(t, store.called) +} + +func TestResolveRIExchange_MapsStoreOutcomes(t *testing.T) { + cases := []struct { + err error + want int + }{ + {fmt.Errorf("%w: wait", config.ErrRIExchangeResolveConflict), 409}, + {fmt.Errorf("%w: other region", config.ErrRIExchangeResolveInvalid), 400}, + } + for _, tc := range cases { + store := &resolveRIStore{MockConfigStore: new(MockConfigStore), err: tc.err} + _, err := riResolveCall(t, apgCaller{perms: riExecute}, store, riResolveFailedBody) + assert.Equal(t, tc.want, clientCode(t, err), tc.err.Error()) + } + + missing := &resolveRIStore{MockConfigStore: new(MockConfigStore), err: config.ErrRIExchangeResolveNotFound} + _, err := riResolveCall(t, apgCaller{perms: riExecute}, missing, riResolveFailedBody) + require.ErrorIs(t, err, errNotFound) + + broken := &resolveRIStore{MockConfigStore: new(MockConfigStore), err: errors.New("db down")} + _, err = riResolveCall(t, apgCaller{perms: riExecute}, broken, riResolveFailedBody) + require.EqualError(t, err, "db down") + _, isClient := IsClientError(err) + assert.False(t, isClient, "a store failure is a 500, not a client error") +} + +// The route is session-authenticated (so the middleware enforces CSRF), never public. +func TestResolveRIExchange_RouteIsAuthUser(t *testing.T) { + r := NewRouter(&Handler{}) + for _, route := range r.routes { + if route.PathPrefix == "/api/ri-exchange/resolve/" { + assert.Equal(t, "POST", route.Method) + assert.Equal(t, AuthUser, route.Auth) + assert.Equal(t, map[string]string{"id": resolveRIID}, r.extractParams(route, "/api/ri-exchange/resolve/"+resolveRIID)) + return + } + } + t.Fatal("POST /api/ri-exchange/resolve/{id} is not registered") +} diff --git a/internal/api/router.go b/internal/api/router.go index 9b5ba7b4e..48fe7a20c 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -321,6 +321,9 @@ func (r *Router) registerRoutes() { {ExactPath: "/api/ri-exchange/history", Method: "GET", Handler: r.getRIExchangeHistoryHandler, Auth: AuthUser}, {PathPrefix: "/api/ri-exchange/approve/", Method: "POST", Handler: r.approveRIExchangeHandler, Auth: AuthPublic}, {PathPrefix: "/api/ri-exchange/reject/", Method: "POST", Handler: r.rejectRIExchangeHandler, Auth: AuthPublic}, + // Operator resolve of a stale processing exchange (#692). AuthUser, so the + // outer middleware enforces the session and CSRF. + {PathPrefix: "/api/ri-exchange/resolve/", Method: "POST", Handler: r.resolveRIExchangeHandler, Auth: AuthUser}, // Commitment Laddering endpoints (flag-gated default-off, issue #1336). // GET returns all per-account ladder configs; PUT inserts or updates one. @@ -908,6 +911,10 @@ func (r *Router) rejectRIExchangeHandler(ctx context.Context, req *events.Lambda return r.h.rejectRIExchange(ctx, params["id"], req.QueryStringParameters["token"]) } +func (r *Router) resolveRIExchangeHandler(ctx context.Context, req *events.LambdaFunctionURLRequest, params map[string]string) (any, error) { + return r.h.resolveRIExchange(ctx, req, params["id"]) +} + // Cloud Account route wrappers. func (r *Router) listAccountsHandler(ctx context.Context, req *events.LambdaFunctionURLRequest, _ map[string]string) (any, error) { diff --git a/internal/config/store_postgres_ri_exchange_resolve.go b/internal/config/store_postgres_ri_exchange_resolve.go new file mode 100644 index 000000000..b1469c1cf --- /dev/null +++ b/internal/config/store_postgres_ri_exchange_resolve.go @@ -0,0 +1,219 @@ +package config + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" +) + +// Outcomes of an operator resolve of a stale processing RI exchange (#692) +// that the store distinguishes. The handler maps them to 404, 400 and 409. +var ( + ErrRIExchangeResolveNotFound = errors.New("ri exchange not found") + ErrRIExchangeResolveInvalid = errors.New("invalid ri exchange resolve request") + ErrRIExchangeResolveConflict = errors.New("ri exchange cannot be resolved") +) + +// RIExchangeResolveStaleAfter is how long a processing RI exchange must have +// been unchanged before an operator may resolve it. Every writer that can +// still settle a processing row (the approval request, the scheduled AUTO +// run) finishes within its own budget: 4m30s on the API path and the task +// timeout (15m default, TASK_TIMEOUT on Fargate) on the scheduled path. The +// margin keeps a late settlement from racing an operator's decision, and +// matches the ladder's fired-tranche threshold. +const RIExchangeResolveStaleAfter = 6 * time.Hour + +// riExchangeResolveNoteMax bounds the note stored in the error column. +const riExchangeResolveNoteMax = 1000 + +// RIExchangeCloudCheck is the operator's statement of where they verified the +// exchange's outcome at AWS before resolving it. +type RIExchangeCloudCheck struct { + AccountID string + Region string + CheckedAt time.Time +} + +// RIExchangeResolve asks to settle a stale processing RI exchange on an +// operator's word. Outcome is StatusCompleted (the exchange happened: the row +// keeps its reserved amount and counts on today's UTC day) or StatusFailed +// (the exchange did not happen: the hold is released). Note is the complete +// text stored in the error column, built by the caller so that a repeat of the +// same request compares equal as a whole. +type RIExchangeResolve struct { + ID string + Outcome string + ExchangeID string + Note string + Check RIExchangeCloudCheck + Actor *string + StaleAfter time.Duration +} + +// RIExchangeResolveResult is the row after the call. Replayed is true when an +// identical earlier request had already resolved it and nothing was written. +type RIExchangeResolveResult struct { + ID string + Status string + PaymentDue string + Note string + Replayed bool +} + +// RIExchangeResolveStore is the store surface of the operator resolve endpoint. +type RIExchangeResolveStore interface { + ResolveStaleRIExchange(ctx context.Context, req RIExchangeResolve) (*RIExchangeResolveResult, error) +} + +var _ RIExchangeResolveStore = (*PostgresStore)(nil) + +// riExchangeResolveRow is the locked row the resolve rules are checked against. +type riExchangeResolveRow struct { + status, accountID, region, exchangeID, note, paymentDue string + updatedAt, dbNow time.Time +} + +// ResolveStaleRIExchange settles a processing RI exchange that no running +// writer can still settle. It locks the row and refuses unless it is still +// processing, has been unchanged for StaleAfter (measured on the database +// clock), and the operator's check names the row's own account and region and +// postdates its last change. Releasing (failed) is also refused when the row +// carries a provider exchange id, since AWS then reported the exchange done. +// A row a late writer completed or failed is never overwritten. +func (s *PostgresStore) ResolveStaleRIExchange(ctx context.Context, req RIExchangeResolve) (*RIExchangeResolveResult, error) { + if err := validateRIExchangeResolve(req); err != nil { + return nil, err + } + res := &RIExchangeResolveResult{ID: req.ID, Status: req.Outcome, Note: req.Note} + err := s.WithTx(ctx, func(tx pgx.Tx) error { + row, err := lockRIExchangeForResolve(ctx, tx, req.ID) + if err != nil { + return err + } + if row.status != "processing" { + return replayRIExchangeResolve(req, row, res) + } + if err := checkRIExchangeResolve(req, row); err != nil { + return err + } + return applyRIExchangeResolve(ctx, tx, req, res) + }) + if err != nil { + return nil, err + } + return res, nil +} + +func validateRIExchangeResolve(req RIExchangeResolve) error { + if req.StaleAfter <= 0 { + return errors.New("ri exchange resolve needs a positive StaleAfter") + } + switch { + case req.Outcome != StatusCompleted && req.Outcome != StatusFailed: + return fmt.Errorf("%w: outcome must be %q or %q", ErrRIExchangeResolveInvalid, StatusCompleted, StatusFailed) + case req.Outcome == StatusFailed && req.ExchangeID != "": + return fmt.Errorf("%w: an exchange id means the exchange happened; resolve it as completed", ErrRIExchangeResolveInvalid) + case req.Note == "" || len(req.Note) > riExchangeResolveNoteMax: + return fmt.Errorf("%w: note must be 1 to %d bytes", ErrRIExchangeResolveInvalid, riExchangeResolveNoteMax) + case !req.Check.complete(): + return fmt.Errorf("%w: the cloud check must name an account, a region and a time", ErrRIExchangeResolveInvalid) + } + return nil +} + +func (c RIExchangeCloudCheck) complete() bool { + return c.AccountID != "" && c.Region != "" && !c.CheckedAt.IsZero() +} + +func lockRIExchangeForResolve(ctx context.Context, tx pgx.Tx, id string) (riExchangeResolveRow, error) { + var row riExchangeResolveRow + err := tx.QueryRow(ctx, ` + SELECT status, account_id, region, COALESCE(exchange_id, ''), COALESCE(error, ''), + payment_due::text, updated_at, NOW() + FROM ri_exchange_history + WHERE id = $1 + FOR UPDATE`, id, + ).Scan(&row.status, &row.accountID, &row.region, &row.exchangeID, &row.note, &row.paymentDue, &row.updatedAt, &row.dbNow) + if errors.Is(err, pgx.ErrNoRows) { + return row, ErrRIExchangeResolveNotFound + } + if err != nil { + return row, fmt.Errorf("failed to lock ri exchange %s: %w", id, err) + } + return row, nil +} + +// replayRIExchangeResolve answers a request that found the row no longer +// processing: success only when it holds exactly what this request would have +// written, else a conflict naming the current status. +func replayRIExchangeResolve(req RIExchangeResolve, row riExchangeResolveRow, res *RIExchangeResolveResult) error { + if row.status == req.Outcome && row.note == req.Note { + res.PaymentDue = row.paymentDue + res.Replayed = true + return nil + } + return fmt.Errorf("%w: exchange is %s, not processing", ErrRIExchangeResolveConflict, row.status) +} + +func checkRIExchangeResolve(req RIExchangeResolve, row riExchangeResolveRow) error { + switch { + case row.dbNow.Sub(row.updatedAt) <= req.StaleAfter: + return fmt.Errorf("%w: the exchange last changed at %s and may still be in flight; wait until it is %s old", + ErrRIExchangeResolveConflict, row.updatedAt.UTC().Format(time.RFC3339), req.StaleAfter) + case row.exchangeID == "": + // No provider exchange id: nothing below applies. + case req.Outcome == StatusFailed: + return fmt.Errorf("%w: the exchange has provider exchange id %q, so it happened; it cannot be released", + ErrRIExchangeResolveConflict, row.exchangeID) + case req.ExchangeID != "" && req.ExchangeID != row.exchangeID: + return fmt.Errorf("%w: exchange_id does not match the recorded provider exchange id", ErrRIExchangeResolveInvalid) + } + return checkRIExchangeCloudCheck(req.Check, row) +} + +// checkRIExchangeCloudCheck requires the operator's check to name the row's own +// account and region and to postdate its last change. +func checkRIExchangeCloudCheck(c RIExchangeCloudCheck, row riExchangeResolveRow) error { + switch { + case c.AccountID != row.accountID: + return fmt.Errorf("%w: checked_account_id is not the exchange's account", ErrRIExchangeResolveInvalid) + case c.Region != row.region: + return fmt.Errorf("%w: checked_region is not the exchange's region", ErrRIExchangeResolveInvalid) + case c.CheckedAt.After(row.dbNow): + return fmt.Errorf("%w: checked_at is in the future", ErrRIExchangeResolveInvalid) + case !c.CheckedAt.After(row.updatedAt): + return fmt.Errorf("%w: checked_at must be after the exchange's last change at %s", + ErrRIExchangeResolveInvalid, row.updatedAt.UTC().Format(time.RFC3339)) + } + return nil +} + +// applyRIExchangeResolve writes the outcome. completed keeps payment_due (the +// reserved ceiling, an upper bound of what AWS accepted) and stamps +// completed_at, so the row counts on today's UTC day like any other settled +// exchange; failed releases the hold. +func applyRIExchangeResolve(ctx context.Context, tx pgx.Tx, req RIExchangeResolve, res *RIExchangeResolveResult) error { + err := tx.QueryRow(ctx, ` + UPDATE ri_exchange_history + SET status = $2::text, + error = $3, + exchange_id = CASE WHEN $4::text = '' THEN exchange_id ELSE $4::text END, + completed_at = CASE WHEN $2::text = 'completed' THEN NOW() ELSE completed_at END, + transitioned_by = $5::uuid, + transitioned_at = NOW(), + updated_at = NOW() + WHERE id = $1 AND status = 'processing' + RETURNING payment_due::text`, + req.ID, req.Outcome, req.Note, req.ExchangeID, req.Actor, + ).Scan(&res.PaymentDue) + if errors.Is(err, pgx.ErrNoRows) { + return fmt.Errorf("%w: exchange changed while resolving", ErrRIExchangeResolveConflict) + } + if err != nil { + return fmt.Errorf("failed to resolve ri exchange %s: %w", req.ID, err) + } + return nil +} diff --git a/internal/config/store_postgres_ri_exchange_resolve_integration_test.go b/internal/config/store_postgres_ri_exchange_resolve_integration_test.go new file mode 100644 index 000000000..695a089de --- /dev/null +++ b/internal/config/store_postgres_ri_exchange_resolve_integration_test.go @@ -0,0 +1,299 @@ +//go:build integration + +package config + +import ( + "errors" + "fmt" + "math/big" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// Real-PostgreSQL coverage for #692: a processing RI exchange left behind by a +// crashed run holds daily-cap headroom until an operator resolves it, and the +// resolve never releases a hold that a running writer may still settle. Uses the +// dedicated issue41_ database guard from newRIExchangeIntegrationEnv. + +const resolveStaleAge = RIExchangeResolveStaleAfter + time.Hour + +func (e *riExchangeIntegrationEnv) resolveReq(id, outcome string, row *RIExchangeRecord) RIExchangeResolve { + return RIExchangeResolve{ + ID: id, Outcome: outcome, Note: "operator:" + outcome + ": verified source RIs in the console", + Check: RIExchangeCloudCheck{AccountID: row.AccountID, Region: row.Region, CheckedAt: time.Now().UTC().Add(-time.Minute)}, + StaleAfter: RIExchangeResolveStaleAfter, + } +} + +func (e *riExchangeIntegrationEnv) status(t *testing.T, id string) (status, note string, completedAt *time.Time) { + t.Helper() + var n *string + require.NoError(t, e.poolA.QueryRow(e.ctx, + `SELECT status, error, completed_at FROM ri_exchange_history WHERE id = $1`, id).Scan(&status, &n, &completedAt)) + if n != nil { + note = *n + } + return status, note, completedAt +} + +func autoCandidate(account int64) *RIExchangeRecord { + return &RIExchangeRecord{AccountID: testAccountID(account), Region: "us-east-1", SourceRIIDs: []string{fmt.Sprintf("ri-resolve-%d", account)}, + SourceInstanceType: "m5.large", SourceCount: 1, TargetOfferingID: "offering-test", TargetInstanceType: "m6i.large", TargetCount: 1, + PaymentDue: "1.000000", Status: "processing", Mode: "auto"} +} + +// The issue's scenario: a crashed run leaves a $1000 hold that blocks every +// later exchange, on every later day. Resolving it as failed after the operator +// verified nothing happened at AWS frees the headroom. +func TestPostgresStoreDB_ResolveStaleRIExchange_FailedReleasesStrandedHold(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + now := time.Now().UTC() + cap := env.capAbove(t, now) + hold := env.processing(t, "auto", testAccountID(692000000001), 1000, now.Add(-resolveStaleAge)) + defer env.release(t, hold.ID) + + next := autoCandidate(692000000002) + _, err := env.storeA.ReserveRIExchange(env.ctx, next, cap, "1000.000000") + require.ErrorIs(t, err, ErrRIExchangeDailyCapExceeded, "the stranded hold must still block before it is resolved") + require.Equal(t, 1, env.spend(t, now.Add(48*time.Hour)).Cmp(usdMicros(999)), "the hold still counts two days later") + + res, err := env.storeA.ResolveStaleRIExchange(env.ctx, env.resolveReq(hold.ID, StatusFailed, hold)) + require.NoError(t, err) + require.False(t, res.Replayed) + require.Equal(t, StatusFailed, res.Status) + status, note, completedAt := env.status(t, hold.ID) + require.Equal(t, StatusFailed, status) + require.Equal(t, res.Note, note) + require.Nil(t, completedAt) + + ceiling, err := env.storeA.ReserveRIExchange(env.ctx, next, cap, "1000.000000") + require.NoError(t, err, "a released hold no longer counts") + require.Equal(t, "1000.000000", ceiling) + env.release(t, next.ID) +} + +// Resolving as completed keeps the reserved amount on the books for today (it +// is an upper bound of what AWS accepted) and lets it drop off tomorrow like +// any settled exchange. +func TestPostgresStoreDB_ResolveStaleRIExchange_CompletedCountsTodayOnly(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + now := time.Now().UTC() + tomorrow := now.Add(24 * time.Hour) + before, beforeTomorrow := env.spend(t, now), env.spend(t, tomorrow) + hold := env.processing(t, "manual", testAccountID(692000000011), 700, now.Add(-resolveStaleAge)) + defer env.release(t, hold.ID) + + req := env.resolveReq(hold.ID, StatusCompleted, hold) + req.ExchangeID = "riex-found-in-console" + res, err := env.storeA.ResolveStaleRIExchange(env.ctx, req) + require.NoError(t, err) + require.Equal(t, "700.000000", res.PaymentDue) + status, _, completedAt := env.status(t, hold.ID) + require.Equal(t, StatusCompleted, status) + require.NotNil(t, completedAt) + var exchangeID string + require.NoError(t, env.poolA.QueryRow(env.ctx, `SELECT exchange_id FROM ri_exchange_history WHERE id = $1`, hold.ID).Scan(&exchangeID)) + require.Equal(t, "riex-found-in-console", exchangeID) + + require.Zero(t, env.spend(t, now).Cmp(new(big.Int).Add(before, usdMicros(700))), "a completed resolve still counts today") + require.Zero(t, env.spend(t, tomorrow).Cmp(beforeTomorrow), "and only today") +} + +// A row that changed inside the in-flight window may still be settled by its +// writer, so it is never resolved, whatever the operator says. +func TestPostgresStoreDB_ResolveStaleRIExchange_RefusesRowThatMayBeInFlight(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + now := time.Now().UTC() + hold := env.processing(t, "auto", testAccountID(692000000021), 10, now.Add(-RIExchangeResolveStaleAfter+time.Minute)) + defer env.release(t, hold.ID) + + for _, outcome := range []string{StatusFailed, StatusCompleted} { + _, err := env.storeA.ResolveStaleRIExchange(env.ctx, env.resolveReq(hold.ID, outcome, hold)) + require.ErrorIs(t, err, ErrRIExchangeResolveConflict, outcome) + status, _, _ := env.status(t, hold.ID) + require.Equal(t, "processing", status) + } +} + +// AWS returned an exchange id, so the exchange happened: it can be settled as +// completed but never released. +func TestPostgresStoreDB_ResolveStaleRIExchange_RefusesToReleaseRowWithExchangeID(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + now := time.Now().UTC() + hold := env.processing(t, "auto", testAccountID(692000000031), 10, now) + _, err := env.poolA.Exec(env.ctx, `UPDATE ri_exchange_history SET exchange_id = 'riex-accepted' WHERE id = $1`, hold.ID) + require.NoError(t, err) + env.setUpdatedAt(t, hold.ID, now.Add(-resolveStaleAge)) + defer env.release(t, hold.ID) + + _, err = env.storeA.ResolveStaleRIExchange(env.ctx, env.resolveReq(hold.ID, StatusFailed, hold)) + require.ErrorIs(t, err, ErrRIExchangeResolveConflict) + status, _, _ := env.status(t, hold.ID) + require.Equal(t, "processing", status) + + mismatch := env.resolveReq(hold.ID, StatusCompleted, hold) + mismatch.ExchangeID = "riex-other" + _, err = env.storeA.ResolveStaleRIExchange(env.ctx, mismatch) + require.ErrorIs(t, err, ErrRIExchangeResolveInvalid) + + _, err = env.storeA.ResolveStaleRIExchange(env.ctx, env.resolveReq(hold.ID, StatusCompleted, hold)) + require.NoError(t, err) +} + +func TestPostgresStoreDB_ResolveStaleRIExchange_ValidatesCloudCheck(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + now := time.Now().UTC() + updated := now.Add(-resolveStaleAge) + hold := env.processing(t, "auto", testAccountID(692000000041), 10, updated) + defer env.release(t, hold.ID) + + cases := map[string]func(*RIExchangeResolve){ + "other account": func(r *RIExchangeResolve) { r.Check.AccountID = testAccountID(692000000042) }, + "other region": func(r *RIExchangeResolve) { r.Check.Region = "eu-west-1" }, + "checked before change": func(r *RIExchangeResolve) { r.Check.CheckedAt = updated.Add(-time.Minute) }, + "checked at the change": func(r *RIExchangeResolve) { r.Check.CheckedAt = updated }, + "checked in the future": func(r *RIExchangeResolve) { r.Check.CheckedAt = now.Add(time.Hour) }, + "failed with exchange id": func(r *RIExchangeResolve) { r.ExchangeID = "riex-x" }, + "unknown outcome": func(r *RIExchangeResolve) { r.Outcome = "canceled" }, + "empty note": func(r *RIExchangeResolve) { r.Note = "" }, + } + for name, mutate := range cases { + t.Run(name, func(t *testing.T) { + req := env.resolveReq(hold.ID, StatusFailed, hold) + mutate(&req) + _, err := env.storeA.ResolveStaleRIExchange(env.ctx, req) + require.ErrorIs(t, err, ErrRIExchangeResolveInvalid) + status, _, _ := env.status(t, hold.ID) + require.Equal(t, "processing", status) + }) + } + + _, err := env.storeA.ResolveStaleRIExchange(env.ctx, env.resolveReq("00000000-0000-4000-8000-000000000692", StatusFailed, hold)) + require.ErrorIs(t, err, ErrRIExchangeResolveNotFound) +} + +// An identical retry answers success without writing; a different request +// against an already-resolved row is a conflict. +func TestPostgresStoreDB_ResolveStaleRIExchange_ReplayAndConflict(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + hold := env.processing(t, "auto", testAccountID(692000000051), 10, time.Now().UTC().Add(-resolveStaleAge)) + defer env.release(t, hold.ID) + + req := env.resolveReq(hold.ID, StatusFailed, hold) + _, err := env.storeA.ResolveStaleRIExchange(env.ctx, req) + require.NoError(t, err) + again, err := env.storeA.ResolveStaleRIExchange(env.ctx, req) + require.NoError(t, err) + require.True(t, again.Replayed) + + _, err = env.storeA.ResolveStaleRIExchange(env.ctx, env.resolveReq(hold.ID, StatusCompleted, hold)) + require.ErrorIs(t, err, ErrRIExchangeResolveConflict) + other := req + other.Note = "operator:failed: a different reason" + _, err = env.storeA.ResolveStaleRIExchange(env.ctx, other) + require.ErrorIs(t, err, ErrRIExchangeResolveConflict) +} + +// A late settlement and an operator release racing on the same row must end +// with the exchange counted, in either lock order: a release that locks first +// is overwritten by the completion queued behind it, and a release queued +// behind the completion sees the row completed and refuses. A third +// transaction holds the row lock so both calls are provably parked, and the +// order they queue in is the order PostgreSQL grants the row lock. +func TestPostgresStoreDB_ResolveStaleRIExchange_RaceWithLateSettlementKeepsHold(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + for _, resolveFirst := range []bool{true, false} { + t.Run(fmt.Sprintf("resolveFirst=%t", resolveFirst), func(t *testing.T) { + hold := env.processing(t, "auto", testAccountID(692000000061), 400, time.Now().UTC().Add(-resolveStaleAge)) + defer env.release(t, hold.ID) + lockTx, err := env.poolA.Begin(env.ctx) + require.NoError(t, err) + defer lockTx.Rollback(env.ctx) + _, err = lockTx.Exec(env.ctx, `SELECT 1 FROM ri_exchange_history WHERE id = $1 FOR UPDATE`, hold.ID) + require.NoError(t, err) + + var wg sync.WaitGroup + var resolveErr, completeErr error + resolve := func() { + defer wg.Done() + _, resolveErr = env.storeA.ResolveStaleRIExchange(env.ctx, env.resolveReq(hold.ID, StatusFailed, hold)) + } + complete := func() { + defer wg.Done() + completeErr = env.storeB.CompleteRIExchangeWithPayment(env.ctx, hold.ID, "riex-late", "400.000000") + } + first, second := resolve, complete + if !resolveFirst { + first, second = complete, resolve + } + wg.Add(2) + go first() + env.waitForLockWaiters(t, 1) + go second() + env.waitForLockWaiters(t, 2) + require.NoError(t, lockTx.Rollback(env.ctx)) + wg.Wait() + + require.NoError(t, completeErr) + if resolveFirst { + require.NoError(t, resolveErr) + } else { + require.ErrorIs(t, resolveErr, ErrRIExchangeResolveConflict, "a release queued behind the settlement must refuse") + } + status, _, completedAt := env.status(t, hold.ID) + require.Equal(t, StatusCompleted, status, "the late settlement must win") + require.NotNil(t, completedAt) + }) + } +} + +// Two operators resolving the same row at once: exactly one decision lands. +func TestPostgresStoreDB_ResolveStaleRIExchange_ConcurrentResolvesOneWins(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + hold := env.processing(t, "auto", testAccountID(692000000071), 10, time.Now().UTC().Add(-resolveStaleAge)) + defer env.release(t, hold.ID) + lockTx, err := env.poolA.Begin(env.ctx) + require.NoError(t, err) + _, err = lockTx.Exec(env.ctx, `SELECT 1 FROM ri_exchange_history WHERE id = $1 FOR UPDATE`, hold.ID) + require.NoError(t, err) + + errs := make([]error, 2) + var wg sync.WaitGroup + for i, run := range []struct { + store *PostgresStore + outcome string + }{{env.storeA, StatusFailed}, {env.storeB, StatusCompleted}} { + wg.Add(1) + go func(i int, store *PostgresStore, outcome string) { + defer wg.Done() + _, errs[i] = store.ResolveStaleRIExchange(env.ctx, env.resolveReq(hold.ID, outcome, hold)) + }(i, run.store, run.outcome) + } + env.waitForLockWaiters(t, 2) + require.NoError(t, lockTx.Rollback(env.ctx)) + wg.Wait() + + wins := 0 + for _, err := range errs { + if err == nil { + wins++ + continue + } + require.True(t, errors.Is(err, ErrRIExchangeResolveConflict), "loser must see a conflict: %v", err) + } + require.Equal(t, 1, wins) +} + +// waitForLockWaiters waits until n backends are blocked on a row lock. +func (e *riExchangeIntegrationEnv) waitForLockWaiters(t *testing.T, n int) { + t.Helper() + require.Eventually(t, func() bool { + var waiting int + err := e.poolA.QueryRow(e.ctx, `SELECT count(*) FROM pg_stat_activity + WHERE datname = current_database() AND wait_event_type = 'Lock'`).Scan(&waiting) + return err == nil && waiting >= n + }, 10*time.Second, 10*time.Millisecond, "both calls must be parked on the row lock") +} diff --git a/internal/server/handler_ri_exchange.go b/internal/server/handler_ri_exchange.go index 73406b5bb..02597f50b 100644 --- a/internal/server/handler_ri_exchange.go +++ b/internal/server/handler_ri_exchange.go @@ -12,6 +12,7 @@ import ( "github.com/LeanerCloud/cloud-commitments-go/pkg/common" "github.com/LeanerCloud/cloud-commitments-go/pkg/exchange" + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" awsprovider "github.com/LeanerCloud/cloud-commitments-go/providers/aws" "github.com/LeanerCloud/cloud-commitments-go/providers/aws/recommendations" ec2svc "github.com/LeanerCloud/cloud-commitments-go/providers/aws/services/ec2" @@ -39,6 +40,9 @@ func (app *Application) handleRIExchangeReshape(ctx context.Context) (*exchange. if err != nil { return nil, fmt.Errorf("failed to load config: %w", err) } + // Before the enabled check: a stranded hold from an earlier run still + // blocks approvals of pending exchanges while automation is off. + app.alarmStaleRIExchanges(ctx) if !cfg.RIExchangeEnabled { log.Println("RI exchange automation is disabled, skipping") return nil, nil @@ -67,6 +71,36 @@ func (app *Application) handleRIExchangeReshape(ctx context.Context) (*exchange. return app.executeRIExchangeReshape(ctx, cfg, clients) } +// riExchangeStaleAlarmAfter is how long a processing RI exchange may go +// unchanged before the scheduled run raises an operator alarm (#692). Every +// writer settles its row well inside this (4m30s on the API path, the task +// timeout on the scheduled path), so a row this old was most likely stranded +// by a crash between the provider call and the ledger write. +const riExchangeStaleAlarmAfter = time.Hour + +// alarmStaleRIExchanges logs an ERROR, which feeds the application error +// alarm, for every processing RI exchange older than riExchangeStaleAlarmAfter. +// Such a row keeps its reserved ceiling counted against the daily cap on every +// later day, and whether the exchange happened at AWS is unknown, so it is +// never settled or released here: an operator verifies it at AWS and resolves +// it via POST /api/ri-exchange/resolve/{id}. It returns how many it reported. +func (app *Application) alarmStaleRIExchanges(ctx context.Context) int { + stale, err := app.Config.GetStaleProcessingExchanges(ctx, riExchangeStaleAlarmAfter) + if err != nil { + logging.Errorf("RI exchange reconciliation: failed to list stale processing exchanges: %v", err) + return 0 + } + for i := range stale { + r := &stale[i] + logging.Errorf("RI exchange reconciliation needed: record_id=%s account_id=%s region=%s source_ri_ids=%v held_usd=%s processing_since=%s: "+ + "the exchange outcome at AWS is unknown and the row keeps holding daily-cap headroom; check the source RIs in AWS, "+ + "then resolve it with POST /api/ri-exchange/resolve/%s (allowed once unchanged for %s)", + r.ID, r.AccountID, r.Region, r.SourceRIIDs, r.PaymentDue, r.UpdatedAt.UTC().Format(time.RFC3339), + r.ID, config.RIExchangeResolveStaleAfter) + } + return len(stale) +} + // convertibleOfferingFinder is the slice of the EC2 client the offering // lookup needs. Satisfied by *ec2svc.Client. type convertibleOfferingFinder interface { diff --git a/internal/server/handler_ri_exchange_stale_test.go b/internal/server/handler_ri_exchange_stale_test.go new file mode 100644 index 000000000..5d62af6d0 --- /dev/null +++ b/internal/server/handler_ri_exchange_stale_test.go @@ -0,0 +1,126 @@ +package server + +import ( + "bytes" + "context" + "errors" + "strings" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/testutil" +) + +// Unit coverage for #692: the scheduled reshape run raises an operator alarm +// for processing RI exchanges stranded by a crashed run. The store is a mock; +// the resolve path is covered against real PostgreSQL in internal/config. + +type lockedBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *lockedBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *lockedBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +// captureLogs routes the package logger to a buffer for the test. Not parallel: +// the logger is process-global. +func captureLogs(t *testing.T) *lockedBuffer { + t.Helper() + out := &lockedBuffer{} + prev := logging.SetOutput(out) + t.Cleanup(func() { logging.SetOutput(prev) }) + return out +} + +func errorLines(logs string) []string { + var lines []string + for _, line := range strings.Split(logs, "\n") { + if strings.Contains(line, "[ERROR]") { + lines = append(lines, line) + } + } + return lines +} + +// A stranded processing row must raise an ERROR naming the row, its hold and +// how to resolve it, even while automation is disabled, and must not be +// settled or released by the scheduled run. +func TestHandleRIExchangeReshape_AlarmsOnStrandedProcessingExchange(t *testing.T) { + logs := captureLogs(t) + var gotOlderThan time.Duration + strandedAt := time.Date(2026, 10, 1, 9, 30, 0, 0, time.UTC) + store := &mockConfigStoreForExchangeStale{ + mockConfigStoreForExchange: mockConfigStoreForExchange{globalConfig: &config.GlobalConfig{RIExchangeEnabled: false}}, + staleFunc: func(_ context.Context, olderThan time.Duration) ([]config.RIExchangeRecord, error) { + gotOlderThan = olderThan + return []config.RIExchangeRecord{{ + ID: "6f1c2a8e-0d4b-4c1e-9a3f-692000000001", Region: "us-east-1", Status: "processing", + SourceRIIDs: []string{"ri-stranded"}, PaymentDue: "1000.000000", UpdatedAt: strandedAt, + }}, nil + }, + } + app := &Application{Config: store} + + result, err := app.handleRIExchangeReshape(testutil.TestContext(t)) + testutil.AssertNoError(t, err) + if result != nil { + t.Fatal("disabled automation must still return a nil result") + } + if gotOlderThan != riExchangeStaleAlarmAfter { + t.Fatalf("stale query threshold = %s, want %s", gotOlderThan, riExchangeStaleAlarmAfter) + } + lines := errorLines(logs.String()) + if len(lines) != 1 { + t.Fatalf("want exactly one ERROR line for the stranded row, got %d:\n%s", len(lines), logs.String()) + } + for _, want := range []string{ + "RI exchange reconciliation needed", "record_id=6f1c2a8e-0d4b-4c1e-9a3f-692000000001", "region=us-east-1", + "source_ri_ids=[ri-stranded]", "held_usd=1000.000000", "processing_since=2026-10-01T09:30:00Z", + "POST /api/ri-exchange/resolve/6f1c2a8e-0d4b-4c1e-9a3f-692000000001", + } { + if !strings.Contains(lines[0], want) { + t.Errorf("alarm line missing %q:\n%s", want, lines[0]) + } + } +} + +func TestAlarmStaleRIExchanges_QueryFailureIsAnError(t *testing.T) { + logs := captureLogs(t) + store := &mockConfigStoreForExchangeStale{ + staleFunc: func(context.Context, time.Duration) ([]config.RIExchangeRecord, error) { + return nil, errors.New("db down") + }, + } + app := &Application{Config: store} + if n := app.alarmStaleRIExchanges(testutil.TestContext(t)); n != 0 { + t.Fatalf("reported %d rows on a failed query", n) + } + lines := errorLines(logs.String()) + if len(lines) != 1 || !strings.Contains(lines[0], "failed to list stale processing exchanges: db down") { + t.Fatalf("a failed stale check must itself raise an ERROR, got:\n%s", logs.String()) + } +} + +func TestAlarmStaleRIExchanges_QuietWithoutStaleRows(t *testing.T) { + logs := captureLogs(t) + app := &Application{Config: &mockConfigStoreForExchangeStale{}} + if n := app.alarmStaleRIExchanges(testutil.TestContext(t)); n != 0 { + t.Fatalf("reported %d rows with none stale", n) + } + if lines := errorLines(logs.String()); len(lines) != 0 { + t.Fatalf("no stale rows must log no ERROR, got:\n%s", logs.String()) + } +}