From accf3ce462507f0b6a78bb68d2b75c988bb86471 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 13 Oct 2026 14:59:13 +0200 Subject: [PATCH 1/2] feat(ladder): operator action to fail an unproven fired tranche Adds POST /api/ladder/tranches/{id}/resolve (outcome failed only; completed answers 501 until the commitment visibility adapter lands, PR5b2). It moves a fired tranche to failed with a required reason, releasing its netting. Authorization: execute:purchases for a real user (admin API key refused) scoped to the tranche's account. Safety: the tranche then the retry-chain tail are locked in the reconciler's and Retry's order; every chain node must be on an allowlist of statuses; an execution that completed blocks it; a node with an unknown outcome (reaped) additionally needs acknowledge_unknown_outcome, a chain unchanged for LadderFiredStaleAfter, and a cloud_check naming the tranche's account and service with a checked_at after the last change. status_reason stores the full request string, so a repeat is a no-op and a different request conflicts. Refs #118 Co-Authored-By: Claude Sonnet 5.5 --- internal/api/handler_ladder_resolve.go | 223 ++++++++++++++++++ internal/api/handler_ladder_resolve_test.go | 202 ++++++++++++++++ .../ladder_resolve_race_integration_test.go | 141 +++++++++++ internal/api/router.go | 5 + .../config/store_postgres_ladder_reconcile.go | 5 +- .../config/store_postgres_ladder_resolve.go | 205 ++++++++++++++++ ...ostgres_ladder_resolve_integration_test.go | 222 +++++++++++++++++ .../store_postgres_ladder_resolve_test.go | 18 ++ 8 files changed, 1019 insertions(+), 2 deletions(-) create mode 100644 internal/api/handler_ladder_resolve.go create mode 100644 internal/api/handler_ladder_resolve_test.go create mode 100644 internal/api/ladder_resolve_race_integration_test.go create mode 100644 internal/config/store_postgres_ladder_resolve.go create mode 100644 internal/config/store_postgres_ladder_resolve_integration_test.go create mode 100644 internal/config/store_postgres_ladder_resolve_test.go diff --git a/internal/api/handler_ladder_resolve.go b/internal/api/handler_ladder_resolve.go new file mode 100644 index 00000000..260b44e9 --- /dev/null +++ b/internal/api/handler_ladder_resolve.go @@ -0,0 +1,223 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "regexp" + "strings" + "time" + "unicode" + "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/LeanerCloud/cloud-commitments-platform/internal/purchase" + "github.com/aws/aws-lambda-go/events" + "github.com/google/uuid" +) + +const ( + ladderResolveReasonMin = 10 + ladderResolveReasonMax = 300 + ladderResolveNoteMax = 200 + ladderResolveBodyMax = 4096 +) + +var ( + ladderResolveAccountRe = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`) + ladderResolveRegionRe = regexp.MustCompile(`^[a-z0-9-]{1,32}$`) +) + +// ladderResolveRequest is the body of POST /api/ladder/tranches/{id}/resolve. +// Only outcome "failed" is implemented; "completed" needs the commitment +// visibility adapter and answers 501 until it exists. +type ladderResolveRequest struct { + Outcome string `json:"outcome"` + Reason string `json:"reason"` + Note string `json:"note"` + Acknowledge bool `json:"acknowledge_unknown_outcome"` + CloudCheck *ladderResolveCloudCheck `json:"cloud_check"` +} + +type ladderResolveCloudCheck struct { + AccountID string `json:"checked_account_id"` + Region string `json:"checked_region"` + Service string `json:"checked_service"` + CheckedAt string `json:"checked_at"` +} + +type ladderResolveResponse struct { + ID string `json:"id"` + Status string `json:"status"` + StatusReason string `json:"status_reason"` + Replayed bool `json:"replayed"` +} + +// ladderResolveStore is the part of the store the endpoint needs. +type ladderResolveStore interface { + LadderTrancheScope(ctx context.Context, id string) (accountID, provider string, err error) + ResolveLadderTrancheFailed(ctx context.Context, req config.LadderResolveFailed) (*config.LadderResolveResult, error) +} + +func (h *Handler) resolveLadderTranche(ctx context.Context, req *events.LambdaFunctionURLRequest, id string) (any, error) { + session, err := h.requirePermission(ctx, req, auth.ActionExecute, auth.ResourcePurchases) + if err != nil { + return nil, err + } + if grantErr := h.requireAutoPurchaseGrant(ctx, req, session); grantErr != nil { + return nil, grantErr + } + if _, parseErr := uuid.Parse(id); parseErr != nil { + return nil, errNotFound + } + body, err := decodeLadderResolve(req.Body) + if err != nil { + return nil, err + } + store, ok := h.config.(ladderResolveStore) + if !ok { + return nil, NewClientError(503, "Ladder resolve storage is unavailable") + } + accountID, err := h.authorizeLadderResolveScope(ctx, session, store, id) + if err != nil { + return nil, err + } + if body.Outcome == "completed" { + return nil, NewClientError(501, "Resolving a tranche as completed is not available yet: it needs the commitment visibility check") + } + return h.applyLadderResolveFailed(ctx, session, store, id, accountID, body) +} + +// authorizeLadderResolveScope checks the caller's account access and +// execute:purchases scope against the tranche's own account and provider, and +// returns the account id. +func (h *Handler) authorizeLadderResolveScope(ctx context.Context, session *Session, store ladderResolveStore, id string) (string, error) { + accountID, provider, err := store.LadderTrancheScope(ctx, id) + if errors.Is(err, config.ErrLadderAmendNotFound) { + return "", errNotFound + } + if err != nil { + return "", err + } + if _, err = h.requireAccountAccess(ctx, session, accountID); err != nil { + return "", err + } + constraints := []auth.PermissionConstraints{{AccountIDs: []string{accountID}, Providers: []string{provider}, StrictScope: true}} + if scopeErr := h.requirePermissionConstraints(ctx, session, auth.ActionExecute, auth.ResourcePurchases, constraints); scopeErr != nil { + return "", scopeErr + } + if common.ProviderType(provider) != common.ProviderAWS { + return "", NewClientError(501, "Ladder execution is currently available only for AWS accounts") + } + return accountID, nil +} + +func (h *Handler) applyLadderResolveFailed(ctx context.Context, session *Session, store ladderResolveStore, id, accountID string, body ladderResolveRequest) (any, error) { + check, err := body.cloudCheck() + if err != nil { + return nil, err + } + res, err := store.ResolveLadderTrancheFailed(ctx, config.LadderResolveFailed{ + TrancheID: id, AccountID: accountID, StatusReason: body.statusReason(check), + Acknowledge: body.Acknowledge, Check: check, Now: time.Now().UTC(), StaleAfter: purchase.LadderFiredStaleAfter, + }) + switch { + case errors.Is(err, config.ErrLadderResolveNotFound): + return nil, errNotFound + case errors.Is(err, config.ErrLadderResolveInvalid): + return nil, NewClientError(400, err.Error()) + case errors.Is(err, config.ErrLadderResolveConflict): + return nil, NewClientError(409, err.Error()) + case err != nil: + return nil, err + } + if !res.Replayed { + logging.Infof("ladder tranche resolved by operator: tranche=%s actor=%q via=%s fired -> failed acknowledged_unknown_outcome=%t status_reason=%q", + id, session.UserID, credentialKind(ctx, session), body.Acknowledge, res.StatusReason) + } + return ladderResolveResponse{ID: res.ID, Status: res.Status, StatusReason: res.StatusReason, Replayed: res.Replayed}, nil +} + +func decodeLadderResolve(raw string) (ladderResolveRequest, error) { + var body ladderResolveRequest + if len(raw) > ladderResolveBodyMax { + return body, NewClientError(400, "Ladder resolve body is too large") + } + decoder := json.NewDecoder(strings.NewReader(raw)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&body); err != nil { + return body, NewClientError(400, "Invalid ladder resolve body") + } + if err := decoder.Decode(new(any)); err != io.EOF { + return body, NewClientError(400, "Expected one ladder resolve body") + } + body.Reason, body.Note = strings.TrimSpace(body.Reason), strings.TrimSpace(body.Note) + if err := body.validate(); err != nil { + return body, err + } + return body, nil +} + +// validate checks the trimmed body's shape; the store decides the rest. +func (body ladderResolveRequest) validate() error { + if body.Outcome != "failed" && body.Outcome != "completed" { + return NewClientError(400, `outcome must be "failed" or "completed"`) + } + if n := utf8.RuneCountInString(body.Reason); n < ladderResolveReasonMin || n > ladderResolveReasonMax { + return NewClientError(400, fmt.Sprintf("reason must be %d to %d characters", ladderResolveReasonMin, ladderResolveReasonMax)) + } + if utf8.RuneCountInString(body.Note) > ladderResolveNoteMax { + return NewClientError(400, fmt.Sprintf("note must be at most %d characters", ladderResolveNoteMax)) + } + if hasControlRune(body.Reason) || hasControlRune(body.Note) { + return NewClientError(400, "reason and note must not contain control characters") + } + if (body.CloudCheck != nil) != body.Acknowledge { + return NewClientError(400, "acknowledge_unknown_outcome and cloud_check must be sent together") + } + return nil +} + +func hasControlRune(s string) bool { + return strings.IndexFunc(s, func(r rune) bool { return unicode.IsControl(r) || r == unicode.ReplacementChar }) >= 0 +} + +// cloudCheck validates the shape of the operator's cloud-side check. Whether it +// names the tranche's own account and service, and postdates the chain, is +// decided by the store under the chain lock. +func (b ladderResolveRequest) cloudCheck() (*config.LadderCloudCheck, error) { + c := b.CloudCheck + if c == nil { + return nil, nil + } + at, err := time.Parse(time.RFC3339, c.CheckedAt) + if err != nil { + return nil, NewClientError(400, "checked_at must be an RFC 3339 timestamp") + } + if !ladderResolveAccountRe.MatchString(c.AccountID) || !ladderResolveRegionRe.MatchString(c.Region) || !ladderResolveRegionRe.MatchString(c.Service) { + return nil, NewClientError(400, "checked_account_id, checked_region and checked_service must be short identifiers") + } + return &config.LadderCloudCheck{AccountID: c.AccountID, Region: c.Region, Service: c.Service, CheckedAt: at.UTC()}, nil +} + +// statusReason is the exact string stored in ladder_tranches.status_reason. 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 ladderResolveRequest) statusReason(check *config.LadderCloudCheck) string { + var s string + if check == nil { + s = fmt.Sprintf("operator:failed:proven: reason=%q", b.Reason) + } else { + s = fmt.Sprintf("operator:failed:unverified: ack=true account=%q region=%q service=%q checked_at=%s reason=%q", + check.AccountID, check.Region, check.Service, check.CheckedAt.Format(time.RFC3339), b.Reason) + } + if b.Note != "" { + s += fmt.Sprintf(" note=%q", b.Note) + } + return s +} diff --git a/internal/api/handler_ladder_resolve_test.go b/internal/api/handler_ladder_resolve_test.go new file mode 100644 index 00000000..d82e5249 --- /dev/null +++ b/internal/api/handler_ladder_resolve_test.go @@ -0,0 +1,202 @@ +package api + +import ( + "context" + "fmt" + "strings" + "testing" + + "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/ladder/tranches/{id}/resolve (platform#118 PR5b). +// The store is a FIXTURE; the locking and chain rules are covered against real +// PostgreSQL in internal/config and ladder_resolve_race_integration_test.go. + +const resolveTrancheID = "0f0f0f0f-1111-4222-8333-444444444444" + +type resolveTestStore struct { + *MockConfigStore + account string + called int + got config.LadderResolveFailed + res *config.LadderResolveResult + err error +} + +func (s *resolveTestStore) LadderTrancheScope(context.Context, string) (string, string, error) { + return s.account, "aws", nil +} + +func (s *resolveTestStore) ResolveLadderTrancheFailed(_ context.Context, r config.LadderResolveFailed) (*config.LadderResolveResult, error) { + s.called++ + s.got = r + if s.res == nil && s.err == nil { + return &config.LadderResolveResult{ID: r.TrancheID, Status: "failed", StatusReason: r.StatusReason}, nil + } + return s.res, s.err +} + +const resolveProvenBody = `{"outcome":"failed","reason":"verified no purchase in the cloud console"}` + +func resolveCall(t *testing.T, caller apgCaller, store *resolveTestStore, body string) error { + t.Helper() + h, req := caller.handlerAndReq(store.MockConfigStore, body) + h.config = store + _, err := h.resolveLadderTranche(context.Background(), req, resolveTrancheID) + return err +} + +func newResolveStore() *resolveTestStore { + m := new(MockConfigStore) + m.GetCloudAccountFn = func(_ context.Context, id string) (*config.CloudAccount, error) { + return &config.CloudAccount{ID: id, Provider: "aws"}, nil + } + return &resolveTestStore{MockConfigStore: m, account: scopedInAccount} +} + +func clientCode(t *testing.T, err error) int { + t.Helper() + ce, ok := IsClientError(err) + require.True(t, ok, "must be a client error, not a 500: %v", err) + return ce.code +} + +func TestLadderResolve_Authorization(t *testing.T) { + constrained := []auth.Permission{{Action: auth.ActionExecute, Resource: auth.ResourcePurchases, + Constraints: &auth.PermissionConstraints{AccountIDs: []string{scopedOutAccount}}}} + tests := []struct { + name string + caller apgCaller + want int // 0 = allowed + }{ + {"execute:purchases user", apgCaller{perms: apgAdminPlus}, 0}, + {"admin without execute:purchases", apgCaller{perms: apgAdminOnly}, 403}, + {"admin API key", apgCaller{key: true}, 403}, + {"execute:purchases scoped to another account", apgCaller{perms: constrained}, 403}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + store := newResolveStore() + err := resolveCall(t, tc.caller, store, resolveProvenBody) + 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 principal whose account list excludes the tranche's account gets the +// enumeration-safe not-found, like the amend endpoint. +func TestLadderResolve_OutOfScopeAccount(t *testing.T) { + store := newResolveStore() + store.account = scopedOutAccount + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", mock.Anything, "tok").Return(&Session{UserID: apgUserID}, nil).Maybe() + mockAuth.grantPermissionsScoped(apgAdminPlus, []string{scopedInAccount}) + h := &Handler{config: store, auth: mockAuth} + req := &events.LambdaFunctionURLRequest{Headers: map[string]string{"Authorization": "Bearer tok"}, Body: resolveProvenBody} + _, err := h.resolveLadderTranche(context.Background(), req, resolveTrancheID) + require.ErrorIs(t, err, errNotFound) + assert.Zero(t, store.called) +} + +func TestLadderResolve_BodyValidation(t *testing.T) { + long := strings.Repeat("x", 301) + check := func(acct, region, svc, at string) string { + return fmt.Sprintf(`"acknowledge_unknown_outcome":true,"cloud_check":{"checked_account_id":%q,"checked_region":%q,"checked_service":%q,"checked_at":%q}`, acct, region, svc, at) + } + for name, body := range map[string]string{ + "empty": ``, + "unknown field": `{"outcome":"failed","reason":"verified no purchase in the cloud console","actor":"x"}`, + "two documents": resolveProvenBody + resolveProvenBody, + "bad outcome": `{"outcome":"voided","reason":"verified no purchase in the cloud console"}`, + "missing reason": `{"outcome":"failed"}`, + "short reason": `{"outcome":"failed","reason":"nope"}`, + "whitespace reason": `{"outcome":"failed","reason":" "}`, + "long reason": `{"outcome":"failed","reason":"` + long + `"}`, + "newline in reason": `{"outcome":"failed","reason":"verified no purchase\nin the console"}`, + "control char in note": `{"outcome":"failed","reason":"verified no purchase in the cloud console","note":"a\u0007b"}`, + "long note": `{"outcome":"failed","reason":"verified no purchase in the cloud console","note":"` + strings.Repeat("n", 201) + `"}`, + "ack without check": `{"outcome":"failed","reason":"verified no purchase in the cloud console","acknowledge_unknown_outcome":true}`, + "check without ack": `{"outcome":"failed","reason":"verified no purchase in the cloud console","cloud_check":{}}`, + "bad checked_at": `{"outcome":"failed","reason":"verified no purchase in the cloud console",` + check("acct-example-1", "us-east-1", "compute-sp", "yesterday") + `}`, + "quote in account": `{"outcome":"failed","reason":"verified no purchase in the cloud console",` + check(`1"2`, "us-east-1", "compute-sp", "2026-01-01T00:00:00Z") + `}`, + "upper-case region": `{"outcome":"failed","reason":"verified no purchase in the cloud console",` + check("acct-example-1", "US East", "compute-sp", "2026-01-01T00:00:00Z") + `}`, + "oversized body": `{"outcome":"failed","reason":"` + strings.Repeat("x", 5000) + `"}`, + } { + t.Run(name, func(t *testing.T) { + store := newResolveStore() + err := resolveCall(t, apgCaller{perms: apgAdminPlus}, store, body) + assert.Equal(t, 400, clientCode(t, err)) + assert.Zero(t, store.called) + }) + } +} + +func TestLadderResolve_CompletedIsNotImplemented(t *testing.T) { + store := newResolveStore() + err := resolveCall(t, apgCaller{perms: apgAdminPlus}, store, `{"outcome":"completed","reason":"verified the commitment in the console"}`) + assert.Equal(t, 501, clientCode(t, err)) + assert.Zero(t, store.called, "completed must not touch the tranche until the visibility check exists") +} + +func TestLadderResolve_StatusReasonIsDeterministic(t *testing.T) { + proven := newResolveStore() + require.NoError(t, resolveCall(t, apgCaller{perms: apgAdminPlus}, proven, resolveProvenBody)) + assert.Equal(t, `operator:failed:proven: reason="verified no purchase in the cloud console"`, proven.got.StatusReason) + assert.False(t, proven.got.Acknowledge) + + unverified := newResolveStore() + body := `{"outcome":"failed","reason":"reaped execution, checked the console","note":"ticket 42","acknowledge_unknown_outcome":true,` + + `"cloud_check":{"checked_account_id":"acct-example-1","checked_region":"us-east-1","checked_service":"compute-sp","checked_at":"2026-10-13T10:00:00+02:00"}}` + require.NoError(t, resolveCall(t, apgCaller{perms: apgAdminPlus}, unverified, body)) + assert.Equal(t, `operator:failed:unverified: ack=true account="acct-example-1" region="us-east-1" service="compute-sp" checked_at=2026-10-13T08:00:00Z reason="reaped execution, checked the console" note="ticket 42"`, + unverified.got.StatusReason) + assert.True(t, unverified.got.Acknowledge) + assert.Equal(t, scopedInAccount, unverified.got.AccountID) +} + +func TestLadderResolve_StoreErrorsMapToStatus(t *testing.T) { + for _, tc := range []struct { + err error + want int + }{ + {config.ErrLadderResolveNotFound, 404}, + {config.ErrLadderResolveInvalid, 400}, + {fmt.Errorf("x: %w", config.ErrLadderResolveConflict), 409}, + } { + store := newResolveStore() + store.err = tc.err + err := resolveCall(t, apgCaller{perms: apgAdminPlus}, store, resolveProvenBody) + if tc.want == 404 { + require.ErrorIs(t, err, errNotFound) + continue + } + assert.Equal(t, tc.want, clientCode(t, err)) + } + store := newResolveStore() + store.err = fmt.Errorf("connection reset") + err := resolveCall(t, apgCaller{perms: apgAdminPlus}, store, resolveProvenBody) + require.Error(t, err) + _, isClient := IsClientError(err) + assert.False(t, isClient, "an unexpected store error is a 500, not a client error") +} + +func TestLadderResolve_BadIDIsNotFound(t *testing.T) { + store := newResolveStore() + h, req := (apgCaller{perms: apgAdminPlus}).handlerAndReq(store.MockConfigStore, resolveProvenBody) + h.config = store + _, err := h.resolveLadderTranche(context.Background(), req, "not-a-uuid") + require.ErrorIs(t, err, errNotFound) +} diff --git a/internal/api/ladder_resolve_race_integration_test.go b/internal/api/ladder_resolve_race_integration_test.go new file mode 100644 index 00000000..6cd05471 --- /dev/null +++ b/internal/api/ladder_resolve_race_integration_test.go @@ -0,0 +1,141 @@ +//go:build integration + +package api + +// Real PostgreSQL: operator resolve x resolve x reconciler x real Retry link on +// one fired tranche (platform#118 PR5b). The resolve takes the tranche row and +// then the chain tail, the order Retry's link and the reconciler use, so +// exactly one writer wins and a failed tranche never has a live successor. + +import ( + "context" + "strings" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/purchase" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestLadderResolve_RacesWithReconcileAndRetry(t *testing.T) { + r := newLadderRetryRig(t) + ctx := context.Background() + h := &Handler{config: r.store} + var accountID string + require.NoError(t, r.store.WithTx(ctx, func(tx pgx.Tx) error { + return tx.QueryRow(ctx, `SELECT cloud_account_id::text FROM ladder_configs WHERE id = $1`, r.cfgID).Scan(&accountID) + })) + + const rounds = 25 + var failedRounds, firedRounds, operatorWins, reconcilerWins int + for round := 0; round < rounds; round++ { + if round > 0 { + r.addFiredTranche(t) + } + resolve := func() { + _, err := r.store.ResolveLadderTrancheFailed(ctx, config.LadderResolveFailed{TrancheID: r.trancheID, AccountID: accountID, + StatusReason: `operator:failed:proven: reason="race"`, Now: time.Now().UTC(), StaleAfter: purchase.LadderFiredStaleAfter}) + if err != nil { + require.ErrorIs(t, err, config.ErrLadderResolveConflict) + } + } + reconcile := func() { + _, err := purchase.ReconcileLadderTranches(ctx, r.store, visibleNever{}, time.Now()) + require.NoError(t, err) + } + retry := func() { + err := h.persistRetrySuccessorTx(ctx, r.failedExec, r.successor(), nil) + if err != nil { + ce, ok := IsClientError(err) + require.True(t, ok, "retry failed unexpectedly: %v", err) + assert.Equal(t, 409, ce.code) + } + } + writers := []func(){resolve, resolve, reconcile, reconcile, retry, retry, retry, retry} + runRaceRound(round, writers[:4], writers[4:]) + + status := r.trancheStatus(t) + linked := r.headLinked(t) + switch status { + case "failed": + failedRounds++ + assert.False(t, linked, "round %d: a failed tranche must not have a live successor", round) + reason := r.trancheReason(t) + if strings.HasPrefix(reason, "operator:") { + operatorWins++ + } else { + assert.Equal(t, "no_submission", reason) + reconcilerWins++ + } + case "fired": + firedRounds++ + assert.True(t, linked, "round %d: a tranche left fired must be the one Retry extended", round) + default: + t.Fatalf("round %d: unexpected tranche status %q", round, status) + } + } + assert.Positive(t, failedRounds, "the resolve/reconcile side never won: the race was not exercised") + assert.Positive(t, firedRounds, "Retry never won: the race was not exercised") + assert.Positive(t, operatorWins, "the operator resolve never won a round") + t.Logf("failed=%d (operator=%d reconciler=%d) fired-by-retry=%d", failedRounds, operatorWins, reconcilerWins, firedRounds) +} + +// runRaceRound runs finishers (resolve, reconcile) against retries. Rounds +// cycle through three schedules so both orders happen: all at once, finishers +// head start, retries head start. +func runRaceRound(round int, finishers, retries []func()) { + var wg sync.WaitGroup + start := make(chan struct{}) + launch := func(fns []func()) { + for _, fn := range fns { + wg.Add(1) + go func() { + defer wg.Done() + <-start + fn() + }() + } + } + switch round % 3 { + case 0: + launch(finishers) + launch(retries) + close(start) + case 1: + launch(finishers) + close(start) + time.Sleep(150 * time.Millisecond) + start = make(chan struct{}) + launch(retries) + close(start) + default: + launch(retries) + close(start) + time.Sleep(150 * time.Millisecond) + start = make(chan struct{}) + launch(finishers) + close(start) + } + wg.Wait() +} + +func (r *ladderRetryRig) headLinked(t *testing.T) bool { + var linked bool + require.NoError(t, r.store.WithTx(context.Background(), func(tx pgx.Tx) error { + return tx.QueryRow(context.Background(), `SELECT retry_execution_id IS NOT NULL FROM purchase_executions WHERE execution_id = $1`, + r.failedExec.ExecutionID).Scan(&linked) + })) + return linked +} + +func (r *ladderRetryRig) trancheReason(t *testing.T) string { + var s string + require.NoError(t, r.store.WithTx(context.Background(), func(tx pgx.Tx) error { + return tx.QueryRow(context.Background(), `SELECT COALESCE(status_reason, '') FROM ladder_tranches WHERE id = $1`, r.trancheID).Scan(&s) + })) + return s +} diff --git a/internal/api/router.go b/internal/api/router.go index ca2d8877..95d3a5eb 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -326,6 +326,7 @@ func (r *Router) registerRoutes() { {ExactPath: "/api/ladder/configs", Method: "PUT", Handler: r.upsertLadderConfigHandler, Auth: AuthUser}, {ExactPath: "/api/ladder/runs", Method: "GET", Handler: r.getLadderTimelineRunsHandler, Auth: AuthUser}, {ExactPath: "/api/ladder/tranches", Method: "GET", Handler: r.getLadderTimelineHandler, Auth: AuthUser}, + {PathPrefix: "/api/ladder/tranches/", PathSuffix: "/resolve", Method: "POST", Handler: r.resolveLadderTrancheHandler, Auth: AuthUser}, {PathPrefix: "/api/ladder/tranches/", Method: "PATCH", Handler: r.amendLadderTrancheHandler, Auth: AuthUser}, // Notification one-click unsubscribe (RFC 8058). AuthPublic: the signed @@ -1024,6 +1025,10 @@ func (r *Router) amendLadderTrancheHandler(ctx context.Context, req *events.Lamb return r.h.amendLadderTranche(ctx, req, params["id"]) } +func (r *Router) resolveLadderTrancheHandler(ctx context.Context, req *events.LambdaFunctionURLRequest, params map[string]string) (any, error) { + return r.h.resolveLadderTranche(ctx, req, params["id"]) +} + func (r *Router) unsubscribeHandler(ctx context.Context, req *events.LambdaFunctionURLRequest, params map[string]string) (any, error) { return r.h.unsubscribeHandler(ctx, req, params) } diff --git a/internal/config/store_postgres_ladder_reconcile.go b/internal/config/store_postgres_ladder_reconcile.go index ce68ec7f..87304839 100644 --- a/internal/config/store_postgres_ladder_reconcile.go +++ b/internal/config/store_postgres_ladder_reconcile.go @@ -27,6 +27,7 @@ type LadderChainNode struct { NoSubmission bool RetryExecutionID *string CreatedAt time.Time + UpdatedAt time.Time } // ListFiredLadderTranches returns fired tranches that have an execution, @@ -99,13 +100,13 @@ func (s *PostgresStore) LockLadderTrancheChainTx(ctx context.Context, tx pgx.Tx, } func readChainNode(ctx context.Context, tx pgx.Tx, executionID string, forUpdate bool) (LadderChainNode, error) { - q := `SELECT execution_id::text, status, no_submission, retry_execution_id::text, created_at + q := `SELECT execution_id::text, status, no_submission, retry_execution_id::text, created_at, updated_at FROM purchase_executions WHERE execution_id = $1` if forUpdate { q += ` FOR UPDATE` } var n LadderChainNode - err := tx.QueryRow(ctx, q, executionID).Scan(&n.ExecutionID, &n.Status, &n.NoSubmission, &n.RetryExecutionID, &n.CreatedAt) + err := tx.QueryRow(ctx, q, executionID).Scan(&n.ExecutionID, &n.Status, &n.NoSubmission, &n.RetryExecutionID, &n.CreatedAt, &n.UpdatedAt) if errors.Is(err, pgx.ErrNoRows) { return n, fmt.Errorf("%w: execution %s", ErrNotFound, executionID) } diff --git a/internal/config/store_postgres_ladder_resolve.go b/internal/config/store_postgres_ladder_resolve.go new file mode 100644 index 00000000..aea265e5 --- /dev/null +++ b/internal/config/store_postgres_ladder_resolve.go @@ -0,0 +1,205 @@ +package config + +import ( + "context" + "errors" + "fmt" + "slices" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/jackc/pgx/v5" +) + +// Outcomes of an operator resolve that the store distinguishes. The handler +// maps them to 404, 400 and 409. +var ( + ErrLadderResolveNotFound = errors.New("ladder tranche not found") + ErrLadderResolveInvalid = errors.New("invalid ladder resolve request") + ErrLadderResolveConflict = errors.New("ladder tranche cannot be resolved") +) + +// ladderStatusReasonMax bounds status_reason; the column has no CHECK. +const ladderStatusReasonMax = 1000 + +// LadderCloudCheck is the operator's statement of where they looked for the +// purchase before declaring an unproven tranche failed. +type LadderCloudCheck struct { + AccountID string + Region string + Service string + CheckedAt time.Time +} + +// LadderResolveFailed asks to move a fired tranche to failed. StatusReason is +// the complete string to store, built by the caller so that a repeat of the +// same request compares equal as a whole. Check is set when the caller +// acknowledges an unknown outcome. +type LadderResolveFailed struct { + TrancheID string + AccountID string + StatusReason string + Acknowledge bool + Check *LadderCloudCheck + Now time.Time + // StaleAfter is how long the chain must have been unchanged before an + // unknown outcome may be declared failed (purchase.LadderFiredStaleAfter). + StaleAfter time.Duration +} + +// LadderResolveResult is the tranche after the call. Replayed is true when an +// identical earlier request had already failed it and nothing was written. +type LadderResolveResult struct { + ID string + Status string + StatusReason string + Replayed bool +} + +// ladderResolvableNodeStatus is the allowlist of execution statuses a chain +// may contain; anything else (pending, approved, running, a future status) can +// still move money, so it blocks the resolve. +func ladderResolvableNodeStatus(status string) bool { + switch status { + case StatusFailed, StatusCanceled, LegacyStatusCanceled, StatusExpired, StatusCompleted, StatusPartiallyCompleted: + return true + } + return false +} + +// ladderNodeProvenUnsent reports whether nothing in the node can have reached +// the provider. +func ladderNodeProvenUnsent(n LadderChainNode) bool { + switch n.Status { + case StatusCanceled, LegacyStatusCanceled, StatusExpired: + return true + case StatusFailed: + return n.NoSubmission + } + return false +} + +// ResolveLadderTrancheFailed moves a fired tranche to failed on an operator's +// word, releasing its netting. It locks the tranche, then the retry-chain +// tail, in the same order as the reconciler and Retry. It refuses while any +// execution of the chain may still run or has bought something, and, when a +// node's outcome is unknown (for example reaped), unless the operator +// acknowledged that and named where they checked. +func (s *PostgresStore) ResolveLadderTrancheFailed(ctx context.Context, req LadderResolveFailed) (*LadderResolveResult, error) { + if len(req.StatusReason) > ladderStatusReasonMax { + return nil, fmt.Errorf("%w: status reason is longer than %d", ErrLadderResolveInvalid, ladderStatusReasonMax) + } + res := &LadderResolveResult{ID: req.TrancheID, Status: string(ladder.TrancheStatusFailed), StatusReason: req.StatusReason} + err := s.WithTx(ctx, func(tx pgx.Tx) error { + chain, err := s.LockLadderTrancheChainTx(ctx, tx, req.TrancheID) + if err != nil { + return err + } + if checkErr := s.checkLadderResolveFailed(ctx, tx, req, chain); checkErr != nil { + return checkErr + } + moved, err := s.TransitionLadderTrancheTx(ctx, tx, req.TrancheID, ladder.TrancheStatusFired, ladder.TrancheStatusFailed, req.StatusReason) + if err != nil { + return err + } + if !moved { + return fmt.Errorf("%w: tranche changed while resolving", ErrLadderResolveConflict) + } + return nil + }) + if errors.Is(err, ErrLadderTrancheNotFired) { + return s.replayLadderResolve(ctx, req, res) + } + if err != nil { + return nil, err + } + return res, nil +} + +// replayLadderResolve answers a request that found the tranche no longer +// fired: 200 only if it is failed with exactly the status reason this request +// would have written, else a conflict naming the current state. +func (s *PostgresStore) replayLadderResolve(ctx context.Context, req LadderResolveFailed, res *LadderResolveResult) (*LadderResolveResult, error) { + var status string + var reason *string + err := s.db.QueryRow(ctx, `SELECT status, status_reason FROM ladder_tranches WHERE id = $1`, req.TrancheID).Scan(&status, &reason) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrLadderResolveNotFound + } + if err != nil { + return nil, fmt.Errorf("read ladder tranche %s: %w", req.TrancheID, err) + } + if status == string(ladder.TrancheStatusFailed) && reason != nil && *reason == req.StatusReason { + res.Replayed = true + return res, nil + } + return nil, fmt.Errorf("%w: tranche is %s", ErrLadderResolveConflict, status) +} + +func (s *PostgresStore) checkLadderResolveFailed(ctx context.Context, tx pgx.Tx, req LadderResolveFailed, chain []LadderChainNode) error { + var layer, account, external string + err := tx.QueryRow(ctx, `SELECT t.layer_type, c.cloud_account_id::text, COALESCE(a.external_id, '') + FROM ladder_tranches t + JOIN ladder_configs c ON c.id = t.config_id + JOIN cloud_accounts a ON a.id = c.cloud_account_id + WHERE t.id = $1`, req.TrancheID).Scan(&layer, &account, &external) + if errors.Is(err, pgx.ErrNoRows) || (err == nil && account != req.AccountID) { + return ErrLadderResolveNotFound + } + if err != nil { + return fmt.Errorf("read ladder tranche scope %s: %w", req.TrancheID, err) + } + if chain[len(chain)-1].RetryExecutionID != nil { + return fmt.Errorf("%w: retry chain is longer than %d executions or changed", ErrLadderResolveConflict, LadderChainMaxDepth) + } + unproven, lastChange, err := scanLadderResolveChain(chain) + if err != nil { + return err + } + if !unproven { + return nil + } + return validateLadderUnverifiedFail(req, layer, []string{account, external}, lastChange) +} + +// scanLadderResolveChain applies the allowlist and completed-node rules and +// reports whether any node's outcome is unknown and when the chain last changed. +func scanLadderResolveChain(chain []LadderChainNode) (unproven bool, lastChange time.Time, err error) { + for _, n := range chain { + if !ladderResolvableNodeStatus(n.Status) { + return false, lastChange, fmt.Errorf("%w: execution %s is %s and may still run", ErrLadderResolveConflict, n.ExecutionID, n.Status) + } + if n.Status == StatusCompleted || n.Status == StatusPartiallyCompleted { + return false, lastChange, fmt.Errorf("%w: execution %s is %s, so a purchase may exist; failing the tranche would re-buy it", ErrLadderResolveConflict, n.ExecutionID, n.Status) + } + unproven = unproven || !ladderNodeProvenUnsent(n) + if n.UpdatedAt.After(lastChange) { + lastChange = n.UpdatedAt + } + } + return unproven, lastChange, nil +} + +func validateLadderUnverifiedFail(req LadderResolveFailed, layer string, accounts []string, lastChange time.Time) error { + if !req.Acknowledge || req.Check == nil { + return fmt.Errorf("%w: an execution of this tranche has an unknown outcome; acknowledge_unknown_outcome and cloud_check are required", ErrLadderResolveConflict) + } + if req.StaleAfter <= 0 { + return errors.New("ladder resolve needs a positive StaleAfter") + } + if req.Now.Sub(lastChange) <= req.StaleAfter { + return fmt.Errorf("%w: the chain changed at %s; wait until it is %s old", ErrLadderResolveConflict, lastChange.UTC().Format(time.RFC3339), req.StaleAfter) + } + c := req.Check + switch { + case !slices.Contains(accounts, c.AccountID): + return fmt.Errorf("%w: checked_account_id is not the tranche's account (id or provider account number)", ErrLadderResolveInvalid) + case c.Service != layer: + return fmt.Errorf("%w: checked_service must be %q", ErrLadderResolveInvalid, layer) + case c.CheckedAt.After(req.Now): + return fmt.Errorf("%w: checked_at is in the future", ErrLadderResolveInvalid) + case !c.CheckedAt.After(lastChange): + return fmt.Errorf("%w: checked_at must be after the last execution change at %s", ErrLadderResolveInvalid, lastChange.UTC().Format(time.RFC3339)) + } + return nil +} diff --git a/internal/config/store_postgres_ladder_resolve_integration_test.go b/internal/config/store_postgres_ladder_resolve_integration_test.go new file mode 100644 index 00000000..b16b9d96 --- /dev/null +++ b/internal/config/store_postgres_ladder_resolve_integration_test.go @@ -0,0 +1,222 @@ +//go:build integration + +package config + +// Real PostgreSQL tests for the operator resolve action (platform#118 PR5b). + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type resolveNode struct { + status string + noSubmission bool + age time.Duration // how long ago the row last changed +} + +type resolveRig struct { + store *PostgresStore + trancheID string + accountID string + external string +} + +// newResolveRig fires one tranche whose execution chain is nodes (head first, +// each linked to the next by retry_execution_id), as the executor and Retry do. +func newResolveRig(ctx context.Context, t *testing.T, store *PostgresStore, nodes []resolveNode) *resolveRig { + t.Helper() + ext := "73" + uuid.NewString()[:10] + cfgID := seedLadderConfigWithExtID(ctx, t, store, ext) + r := &resolveRig{store: store, external: ext} + require.NoError(t, store.db.QueryRow(ctx, `SELECT cloud_account_id::text FROM ladder_configs WHERE id = $1`, cfgID).Scan(&r.accountID)) + tr := lifecycleTranche(cfgID, 5, ladder.TrancheStatusScheduled, time.Now().Add(-time.Hour)) + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) + r.trancheID = tr.ID + var ids []string + _, err := store.ClaimDueLadderTranches(ctx, cfgID, time.Now(), 1, func(ctx context.Context, tx pgx.Tx, _ LadderTrancheDB) (string, error) { + id, err := insertResolveNode(ctx, tx, tr.ID, nodes[0]) + ids = append(ids, id) + return id, err + }) + require.NoError(t, err) + for _, n := range nodes[1:] { + require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { + id, err := insertResolveNode(ctx, tx, tr.ID, n) + if err != nil { + return err + } + _, err = tx.Exec(ctx, `UPDATE purchase_executions SET retry_execution_id = $2 WHERE execution_id = $1`, ids[len(ids)-1], id) + ids = append(ids, id) + return err + })) + } + for i, n := range nodes { // age rows after linking, with the updated_at trigger off + _, err := store.db.Exec(ctx, `ALTER TABLE purchase_executions DISABLE TRIGGER update_purchase_executions_updated_at`) + require.NoError(t, err) + _, err = store.db.Exec(ctx, `UPDATE purchase_executions SET updated_at = now() - make_interval(secs => $2) WHERE execution_id = $1`, ids[i], n.age.Seconds()) + require.NoError(t, err) + _, err = store.db.Exec(ctx, `ALTER TABLE purchase_executions ENABLE TRIGGER update_purchase_executions_updated_at`) + require.NoError(t, err) + } + return r +} + +func insertResolveNode(ctx context.Context, tx pgx.Tx, key string, n resolveNode) (string, error) { + var id string + err := tx.QueryRow(ctx, `INSERT INTO purchase_executions (scheduled_date, status, no_submission, error, idempotency_key) + VALUES (NOW(), $1, $2, '', $3) RETURNING execution_id::text`, n.status, n.noSubmission, key).Scan(&id) + return id, err +} + +func (r *resolveRig) req(reason string, ack *LadderCloudCheck) LadderResolveFailed { + return LadderResolveFailed{TrancheID: r.trancheID, AccountID: r.accountID, StatusReason: reason, + Acknowledge: ack != nil, Check: ack, Now: time.Now().UTC(), StaleAfter: 6 * time.Hour} +} + +func (r *resolveRig) check(at time.Time) *LadderCloudCheck { + return &LadderCloudCheck{AccountID: r.accountID, Region: "us-east-1", Service: string(ladder.LayerComputeSP), CheckedAt: at} +} + +func TestResolveLadderTrancheFailed_ProvenChain(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + for name, nodes := range map[string][]resolveNode{ + "no_submission head": {{"failed", true, time.Minute}}, + "canceled after proof": {{"failed", true, time.Hour}, {"canceled", false, time.Minute}}, + "legacy spelling": {{LegacyStatusCanceled, false, time.Minute}}, + "expired": {{"expired", false, time.Minute}}, + } { + t.Run(name, func(t *testing.T) { + r := newResolveRig(ctx, t, store, nodes) + res, err := r.store.ResolveLadderTrancheFailed(ctx, r.req("operator:failed:proven: reason=x", nil)) + require.NoError(t, err) + assert.False(t, res.Replayed) + assert.Equal(t, ladder.TrancheStatusFailed, trancheStatus(ctx, t, store, r.trancheID)) + var reason string + require.NoError(t, store.db.QueryRow(ctx, `SELECT status_reason FROM ladder_tranches WHERE id = $1`, r.trancheID).Scan(&reason)) + assert.Equal(t, "operator:failed:proven: reason=x", reason) + }) + } +} + +// Anything that may still run, may have bought, or is not a known status blocks +// the resolve, and the tranche stays fired. +func TestResolveLadderTrancheFailed_Refusals(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + blocked := map[string][]resolveNode{} + for _, s := range []string{"pending", "notified", "scheduled", "approved", "running"} { + blocked["live "+s+" tail"] = []resolveNode{{"failed", true, 8 * time.Hour}, {s, false, 7 * time.Hour}} + blocked["live "+s+" head"] = []resolveNode{{s, false, 7 * time.Hour}} + } + blocked["completed head"] = []resolveNode{{"completed", false, 7 * time.Hour}} + blocked["partially_completed"] = []resolveNode{{"partially_completed", false, 7 * time.Hour}} + blocked["completed then failed retry"] = []resolveNode{{"completed", false, 8 * time.Hour}, {"failed", true, 7 * time.Hour}} + long := make([]resolveNode, LadderChainMaxDepth+1) + for i := range long { + long[i] = resolveNode{"failed", true, 8 * time.Hour} + } + blocked["chain longer than the depth cap"] = long + for name, nodes := range blocked { + t.Run(name, func(t *testing.T) { + r := newResolveRig(ctx, t, store, nodes) + c := r.check(time.Now().Add(-time.Minute)) + _, err := r.store.ResolveLadderTrancheFailed(ctx, r.req("operator:failed:unverified: x", c)) + require.ErrorIs(t, err, ErrLadderResolveConflict) + assert.Equal(t, ladder.TrancheStatusFired, trancheStatus(ctx, t, store, r.trancheID)) + }) + } +} + +func TestResolveLadderTrancheFailed_UnknownOutcome(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + old := []resolveNode{{"failed", false, 7 * time.Hour}} // reaped: no proof + cases := []struct { + name string + nodes []resolveNode + tweak func(*LadderResolveFailed, *resolveRig) + want error + }{ + {"no acknowledgement", old, func(q *LadderResolveFailed, _ *resolveRig) { q.Acknowledge, q.Check = false, nil }, ErrLadderResolveConflict}, + {"acknowledged without check", old, func(q *LadderResolveFailed, _ *resolveRig) { q.Check = nil }, ErrLadderResolveConflict}, + {"changed 5m ago though created long ago", []resolveNode{{"failed", false, 5 * time.Minute}}, nil, ErrLadderResolveConflict}, + {"unproven predecessor with proven tail", []resolveNode{{"failed", false, 8 * time.Hour}, {"failed", true, 7 * time.Hour}}, nil, nil}, + {"wrong account", old, func(q *LadderResolveFailed, _ *resolveRig) { q.Check.AccountID = "acct-other" }, ErrLadderResolveInvalid}, + {"wrong service", old, func(q *LadderResolveFailed, _ *resolveRig) { q.Check.Service = "rds" }, ErrLadderResolveInvalid}, + {"checked in the future", old, func(q *LadderResolveFailed, _ *resolveRig) { q.Check.CheckedAt = q.Now.Add(time.Hour) }, ErrLadderResolveInvalid}, + {"checked before the last change", old, func(q *LadderResolveFailed, _ *resolveRig) { q.Check.CheckedAt = q.Now.Add(-8 * time.Hour) }, ErrLadderResolveInvalid}, + {"valid", old, nil, nil}, + {"valid with provider account number", old, func(q *LadderResolveFailed, r *resolveRig) { q.Check.AccountID = r.external }, nil}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + r := newResolveRig(ctx, t, store, tc.nodes) + q := r.req("operator:failed:unverified: x", r.check(time.Now().Add(-time.Minute))) + if tc.tweak != nil { + tc.tweak(&q, r) + } + _, err := r.store.ResolveLadderTrancheFailed(ctx, q) + if tc.want == nil { + require.NoError(t, err) + assert.Equal(t, ladder.TrancheStatusFailed, trancheStatus(ctx, t, store, r.trancheID)) + return + } + require.ErrorIs(t, err, tc.want) + assert.Equal(t, ladder.TrancheStatusFired, trancheStatus(ctx, t, store, r.trancheID)) + }) + } +} + +func TestResolveLadderTrancheFailed_ReplayAndScope(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + r := newResolveRig(ctx, t, store, []resolveNode{{"failed", true, time.Minute}}) + + t.Run("another account's tranche is not found", func(t *testing.T) { + q := r.req("operator:failed:proven: reason=x", nil) + q.AccountID = uuid.NewString() + _, err := store.ResolveLadderTrancheFailed(ctx, q) + require.ErrorIs(t, err, ErrLadderResolveNotFound) + assert.Equal(t, ladder.TrancheStatusFired, trancheStatus(ctx, t, store, r.trancheID)) + }) + t.Run("unknown tranche", func(t *testing.T) { + q := r.req("operator:failed:proven: reason=x", nil) + q.TrancheID = uuid.NewString() + _, err := store.ResolveLadderTrancheFailed(ctx, q) + require.ErrorIs(t, err, ErrLadderResolveNotFound) + }) + t.Run("oversized status reason", func(t *testing.T) { + _, err := store.ResolveLadderTrancheFailed(ctx, r.req(string(make([]byte, ladderStatusReasonMax+1)), nil)) + require.ErrorIs(t, err, ErrLadderResolveInvalid) + }) + t.Run("repeat is a no-op, a different reason conflicts", func(t *testing.T) { + first, err := store.ResolveLadderTrancheFailed(ctx, r.req("operator:failed:proven: reason=a", nil)) + require.NoError(t, err) + assert.False(t, first.Replayed) + again, err := store.ResolveLadderTrancheFailed(ctx, r.req("operator:failed:proven: reason=a", nil)) + require.NoError(t, err) + assert.True(t, again.Replayed) + _, err = store.ResolveLadderTrancheFailed(ctx, r.req("operator:failed:proven: reason=b", nil)) + require.ErrorIs(t, err, ErrLadderResolveConflict) + var reason string + require.NoError(t, store.db.QueryRow(ctx, `SELECT status_reason FROM ladder_tranches WHERE id = $1`, r.trancheID).Scan(&reason)) + assert.Equal(t, "operator:failed:proven: reason=a", reason) + }) + t.Run("a tranche completed elsewhere conflicts", func(t *testing.T) { + done := newResolveRig(ctx, t, store, []resolveNode{{"completed", false, time.Minute}}) + moved, err := store.TransitionLadderTranche(ctx, done.trancheID, ladder.TrancheStatusFired, ladder.TrancheStatusCompleted, "") + require.NoError(t, err) + require.True(t, moved) + _, err = store.ResolveLadderTrancheFailed(ctx, done.req("operator:failed:proven: reason=x", nil)) + require.ErrorIs(t, err, ErrLadderResolveConflict) + }) +} diff --git a/internal/config/store_postgres_ladder_resolve_test.go b/internal/config/store_postgres_ladder_resolve_test.go new file mode 100644 index 00000000..ba4c4287 --- /dev/null +++ b/internal/config/store_postgres_ladder_resolve_test.go @@ -0,0 +1,18 @@ +package config + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +// The chain-node allowlist fails closed: a live status or one added later is +// not resolvable (the status CHECK keeps such rows out of the database tests). +func TestLadderResolvableNodeStatus(t *testing.T) { + for _, s := range []string{StatusFailed, StatusCanceled, LegacyStatusCanceled, StatusExpired, StatusCompleted, StatusPartiallyCompleted} { + assert.True(t, ladderResolvableNodeStatus(s), s) + } + for _, s := range []string{"pending", "notified", "scheduled", "approved", "running", "", "future-status"} { + assert.False(t, ladderResolvableNodeStatus(s), s) + } +} From 7946f8e3ac47c9429ad047596606a73baf717c9e Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 13 Oct 2026 15:17:17 +0200 Subject: [PATCH 2/2] test(ladder): remove the data race in the resolve race test CI runs integration tests with -race, which flagged the gate channel the test reassigned while goroutines read it. Each phase now gets its own channel. Refs #118 Co-Authored-By: Claude Sonnet 5.5 --- .../ladder_resolve_race_integration_test.go | 29 +++++++++---------- 1 file changed, 13 insertions(+), 16 deletions(-) diff --git a/internal/api/ladder_resolve_race_integration_test.go b/internal/api/ladder_resolve_race_integration_test.go index 6cd05471..25bd5ec3 100644 --- a/internal/api/ladder_resolve_race_integration_test.go +++ b/internal/api/ladder_resolve_race_integration_test.go @@ -89,36 +89,33 @@ func TestLadderResolve_RacesWithReconcileAndRetry(t *testing.T) { // head start, retries head start. func runRaceRound(round int, finishers, retries []func()) { var wg sync.WaitGroup - start := make(chan struct{}) - launch := func(fns []func()) { + // launch parks each function on its own gate channel (never reassigned, so + // no goroutine reads a variable another one writes) and returns the opener. + launch := func(fns []func()) func() { + gate := make(chan struct{}) for _, fn := range fns { wg.Add(1) go func() { defer wg.Done() - <-start + <-gate fn() }() } + return func() { close(gate) } } switch round % 3 { case 0: - launch(finishers) - launch(retries) - close(start) + openFinishers, openRetries := launch(finishers), launch(retries) + openFinishers() + openRetries() case 1: - launch(finishers) - close(start) + launch(finishers)() time.Sleep(150 * time.Millisecond) - start = make(chan struct{}) - launch(retries) - close(start) + launch(retries)() default: - launch(retries) - close(start) + launch(retries)() time.Sleep(150 * time.Millisecond) - start = make(chan struct{}) - launch(finishers) - close(start) + launch(finishers)() } wg.Wait() }