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()) + } +}