diff --git a/internal/api/handler_plans.go b/internal/api/handler_plans.go index 46784cd7..3d9cb25b 100644 --- a/internal/api/handler_plans.go +++ b/internal/api/handler_plans.go @@ -434,6 +434,10 @@ func (h *Handler) deletePlan(ctx context.Context, req *events.LambdaFunctionURLR } if err := h.config.DeletePurchasePlan(ctx, planID); err != nil { + if errors.Is(err, config.ErrManagedPlanLiveExecutions) { + return nil, planEditorRefusal(http.StatusConflict, codeManagedPlanLiveExecutions, + "this plan still has active or unsettled purchases; cancel or revoke them first") + } return nil, err } diff --git a/internal/api/handler_plans_managed_add_accounts_integration_test.go b/internal/api/handler_plans_managed_add_accounts_integration_test.go index 21cdaa98..aaa5d8a2 100644 --- a/internal/api/handler_plans_managed_add_accounts_integration_test.go +++ b/internal/api/handler_plans_managed_add_accounts_integration_test.go @@ -13,6 +13,7 @@ import ( "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/jackc/pgx/v5" "github.com/stretchr/testify/require" ) @@ -44,6 +45,27 @@ func (f *createConcurrencyFixture) extraAccount(ctx context.Context, t *testing. return acct.ID } +// holdPlanLock stands in for the editor command: an open transaction holding +// the plan row FOR UPDATE, rolled back at test end unless the test commits it. +func (f *createConcurrencyFixture) holdPlanLock(ctx context.Context, t *testing.T) pgx.Tx { + t.Helper() + command, err := f.pool.Begin(ctx) + require.NoError(t, err) + t.Cleanup(func() { _ = command.Rollback(ctx) }) + _, err = command.Exec(ctx, `SELECT 1 FROM purchase_plans WHERE id = $1 FOR UPDATE`, f.planID) + require.NoError(t, err) + return command +} + +func awaitLockWait(ctx context.Context, t *testing.T, f *createConcurrencyFixture) { + t.Helper() + require.Eventually(t, func() bool { + var n int + require.NoError(t, f.pool.QueryRow(ctx, `SELECT count(*) FROM pg_stat_activity WHERE wait_event_type = 'Lock'`).Scan(&n)) + return n > 0 + }, 15*time.Second, 50*time.Millisecond, "writer must wait on the plan lock") +} + const ( execCount = `SELECT count(*) FROM purchase_executions WHERE plan_id = $1` acctCount = `SELECT count(*) FROM plan_accounts WHERE plan_id = $1` @@ -130,21 +152,12 @@ func TestManagedMarkerCommittedUnderLockRefusesWaitingWriters(t *testing.T) { t.Run(name, func(t *testing.T) { f := newCreateConcurrencyFixture(ctx, t) o := other(f) - command, err := f.pool.Begin(ctx) - require.NoError(t, err) - defer func() { _ = command.Rollback(ctx) }() - _, err = command.Exec(ctx, `SELECT 1 FROM purchase_plans WHERE id = $1 FOR UPDATE`, f.planID) - require.NoError(t, err) - + command := f.holdPlanLock(ctx, t) done := make(chan error, 1) go func() { done <- write(f, o) }() - require.Eventually(t, func() bool { - var n int - require.NoError(t, f.pool.QueryRow(ctx, `SELECT count(*) FROM pg_stat_activity WHERE wait_event_type = 'Lock'`).Scan(&n)) - return n > 0 - }, 15*time.Second, 50*time.Millisecond, "writer must wait on the plan lock") + awaitLockWait(ctx, t, f) - _, err = command.Exec(ctx, `UPDATE purchase_plans SET ramp_schedule = ramp_schedule || '{"dates_managed":true}' WHERE id = $1`, f.planID) + _, err := command.Exec(ctx, `UPDATE purchase_plans SET ramp_schedule = ramp_schedule || '{"dates_managed":true}' WHERE id = $1`, f.planID) require.NoError(t, err) require.NoError(t, command.Commit(ctx)) requireEditorRefusal(t, <-done) diff --git a/internal/api/handler_plans_managed_delete_resume_integration_test.go b/internal/api/handler_plans_managed_delete_resume_integration_test.go new file mode 100644 index 00000000..7ae49c97 --- /dev/null +++ b/internal/api/handler_plans_managed_delete_resume_integration_test.go @@ -0,0 +1,181 @@ +//go:build integration + +package api + +import ( + "context" + "errors" + "net/http" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/google/uuid" + "github.com/stretchr/testify/require" +) + +func (f *createConcurrencyFixture) saveRow(ctx context.Context, t *testing.T, status string, noSubmission bool, scheduled time.Time, recs bool) string { + t.Helper() + row := &config.PurchaseExecution{ExecutionID: uuid.NewString(), PlanID: f.planID, Status: status, StepNumber: 3, ScheduledDate: scheduled, NoSubmission: noSubmission} + if recs { + row.Recommendations = []config.RecommendationRecord{{ID: "r1", Provider: "azure", Service: "compute", Selected: true}} + } + require.NoError(t, f.store.SavePurchaseExecution(ctx, row)) + return row.ExecutionID +} + +func (f *createConcurrencyFixture) deletePlan(ctx context.Context, h *Handler) error { + _, err := h.deletePlan(ctx, managedGuardRequest(""), f.planID) + return err +} + +func requireLiveExecutionsRefusal(t *testing.T, err error) { + t.Helper() + var ce *clientError + require.True(t, errors.As(err, &ce), "want a client error, got %v", err) + require.Equal(t, http.StatusConflict, ce.code) + require.Equal(t, codeManagedPlanLiveExecutions, ce.details["code"]) +} + +func TestDeleteManagedPlanByExecutionStatus(t *testing.T) { + ctx := context.Background() + cases := []struct { + status string + noSubmission bool + blocks bool + }{ + {"pending", false, true}, {"notified", false, true}, {"scheduled", false, true}, {"approved", false, true}, + {"running", false, true}, {"paused", false, true}, {"failed", false, true}, + {"completed", false, false}, {"partially_completed", false, false}, {"canceled", false, false}, + {"expired", false, false}, {"failed", true, false}, + } + for _, tc := range cases { + name := tc.status + if tc.noSubmission { + name += "_no_submission" + } + t.Run(name, func(t *testing.T) { + f := newCreateConcurrencyFixture(ctx, t) + setPlanManaged(ctx, t, f, true) + id := f.saveRow(ctx, t, tc.status, tc.noSubmission, time.Now(), false) + err := f.deletePlan(ctx, f.handler) + var planID *string + require.NoError(t, f.pool.QueryRow(ctx, `SELECT plan_id::text FROM purchase_executions WHERE execution_id = $1`, id).Scan(&planID)) + if tc.blocks { + requireLiveExecutionsRefusal(t, err) + require.NotNil(t, planID, "a refused delete must not detach the root") + require.Equal(t, 1, f.count(ctx, t, `SELECT count(*) FROM purchase_plans WHERE id = $1`)) + return + } + require.NoError(t, err) + require.Nil(t, planID, "settled history detaches as before") + }) + } + t.Run("unmanaged plan with a live row deletes as before", func(t *testing.T) { + f := newCreateConcurrencyFixture(ctx, t) + f.saveRow(ctx, t, "pending", false, time.Now(), false) + require.NoError(t, f.deletePlan(ctx, f.handler)) + }) +} + +func TestDeleteManagedPlanWaitsForCommandAndSeesItsRoot(t *testing.T) { + ctx := context.Background() + f := newCreateConcurrencyFixture(ctx, t) + command := f.holdPlanLock(ctx, t) + done := make(chan error, 1) + go func() { done <- f.deletePlan(ctx, f.handler) }() + awaitLockWait(ctx, t, f) + + id := uuid.NewString() + _, err := command.Exec(ctx, `INSERT INTO purchase_executions (execution_id, plan_id, status, step_number, scheduled_date) + VALUES ($1, $2, 'pending', 1, now())`, id, f.planID) + require.NoError(t, err) + _, err = command.Exec(ctx, `UPDATE purchase_plans SET ramp_schedule = ramp_schedule || '{"dates_managed":true}' WHERE id = $1`, f.planID) + require.NoError(t, err) + require.NoError(t, command.Commit(ctx)) + + requireLiveExecutionsRefusal(t, <-done) + require.Equal(t, 1, f.count(ctx, t, `SELECT count(*) FROM purchase_executions WHERE plan_id = $1`)) +} + +func TestRescheduleAwaitingExecutionOnManagedPlan(t *testing.T) { + ctx := context.Background() + f := newCreateConcurrencyFixture(ctx, t) + orig := time.Now().Add(-time.Hour).UTC().Truncate(time.Second) + id := f.saveRow(ctx, t, "pending", false, orig, true) + moved := time.Now().Add(48 * time.Hour).UTC().Truncate(time.Second) + + ok, err := f.store.RescheduleAwaitingExecution(ctx, id, moved) + require.NoError(t, err) + require.True(t, ok, "unmanaged plan reschedules as before") + + setPlanManaged(ctx, t, f, true) + ok, err = f.store.RescheduleAwaitingExecution(ctx, id, orig) + require.ErrorIs(t, err, config.ErrEditorReconciliationRequired) + require.False(t, ok) + row, err := f.store.GetExecutionByID(ctx, id) + require.NoError(t, err) + require.True(t, row.ScheduledDate.Equal(moved), "refused reschedule leaves the date") +} + +func TestRescheduleWaitsForCommandThenRefuses(t *testing.T) { + ctx := context.Background() + f := newCreateConcurrencyFixture(ctx, t) + orig := time.Now().Add(-time.Hour).UTC().Truncate(time.Second) + id := f.saveRow(ctx, t, "pending", false, orig, true) + command := f.holdPlanLock(ctx, t) + done := make(chan error, 1) + go func() { + _, err := f.store.RescheduleAwaitingExecution(ctx, id, time.Now().Add(48*time.Hour)) + done <- err + }() + awaitLockWait(ctx, t, f) + _, err := command.Exec(ctx, `UPDATE purchase_plans SET ramp_schedule = ramp_schedule || '{"dates_managed":true}' WHERE id = $1`, f.planID) + require.NoError(t, err) + require.NoError(t, command.Commit(ctx)) + + require.ErrorIs(t, <-done, config.ErrEditorReconciliationRequired) + row, err := f.store.GetExecutionByID(ctx, id) + require.NoError(t, err) + require.True(t, row.ScheduledDate.Equal(orig)) +} + +func TestResumeOnManagedPlanRefusesBeforeTransition(t *testing.T) { + ctx := context.Background() + f := newCreateConcurrencyFixture(ctx, t) + setPlanManaged(ctx, t, f, true) + id := f.saveRow(ctx, t, "paused", false, time.Now().Add(-time.Hour), true) + + _, err := f.handler.resumePlannedPurchase(ctx, managedGuardRequest(""), id) + requireEditorRefusal(t, err) + row, err := f.store.GetExecutionByID(ctx, id) + require.NoError(t, err) + require.Equal(t, "paused", row.Status) +} + +func TestDeleteAndResumeAuthorizationPrecedeManagedGuard(t *testing.T) { + ctx := context.Background() + f := newCreateConcurrencyFixture(ctx, t) + setPlanManaged(ctx, t, f, true) + id := f.saveRow(ctx, t, "paused", false, time.Now().Add(-time.Hour), true) + f.saveRow(ctx, t, "pending", false, time.Now(), false) + view := []auth.Permission{{Action: auth.ActionView, Resource: auth.ResourcePlans}} + admin := []auth.Permission{{Action: auth.ActionAdmin, Resource: auth.ResourceAll}} + + forbidden := limitedHandler(f, view, nil) + scoped := limitedHandler(f, admin, []string{"unrelated-account"}) + _, resumeErr := forbidden.resumePlannedPurchase(ctx, managedGuardRequest(""), id) + for name, got := range map[string]error{"delete": f.deletePlan(ctx, forbidden), "resume": resumeErr} { + var ce *clientError + require.True(t, errors.As(got, &ce), "%s: %v", name, got) + require.Equal(t, http.StatusForbidden, ce.code, name) + } + _, scopedResumeErr := scoped.resumePlannedPurchase(ctx, managedGuardRequest(""), id) + for name, got := range map[string]error{"delete": f.deletePlan(ctx, scoped), "resume": scopedResumeErr} { + require.True(t, IsNotFoundError(got), "scoped %s must look like not found, got %v", name, got) + } + row, err := f.store.GetExecutionByID(ctx, id) + require.NoError(t, err) + require.Equal(t, "paused", row.Status) +} diff --git a/internal/api/handler_purchases.go b/internal/api/handler_purchases.go index eba8adb8..8543980c 100644 --- a/internal/api/handler_purchases.go +++ b/internal/api/handler_purchases.go @@ -278,6 +278,9 @@ func (h *Handler) requireResumeVetoWindow(ctx context.Context, executionID strin if err != nil { return fmt.Errorf("failed to read plan %s before resuming execution %s: %w", row.PlanID, executionID, err) } + if plan.RampSchedule.DatesManaged { + return editorReconciliationError() + } if vetoErr := plan.RequireVetoWindow(); vetoErr != nil { return NewClientError(409, fmt.Sprintf("execution %s cannot be resumed: %v", executionID, vetoErr)) } @@ -306,6 +309,9 @@ func (h *Handler) pushPastScheduledDate(ctx context.Context, row *config.Purchas // approve landing after the resume must not be overwritten. A lost // condition is a benign no-op (the row is no longer awaiting approval). if _, err := h.config.RescheduleAwaitingExecution(ctx, row.ExecutionID, now.AddDate(0, 0, plan.NotificationDaysBefore)); err != nil { + if errors.Is(err, config.ErrEditorReconciliationRequired) { + return editorReconciliationError() + } return fmt.Errorf("failed to reschedule resumed execution %s: %w", row.ExecutionID, err) } return nil diff --git a/internal/api/openapi.yaml b/internal/api/openapi.yaml index 517b21e4..0870de14 100644 --- a/internal/api/openapi.yaml +++ b/internal/api/openapi.yaml @@ -362,11 +362,11 @@ paths: '404': $ref: '#/components/responses/NotFound' '409': - description: The plan changed concurrently, or the request changes the ramp schedule of a plan that has already completed a step. Keep the schedule, or create a new plan. + description: The plan changed concurrently, the request changes the ramp schedule of a plan that has already completed a step (keep the schedule, or create a new plan), or the plan is managed by the schedule editor and the request changes what only the editor may (code editor_reconciliation_required). content: application/json: schema: - $ref: '#/components/schemas/Error' + $ref: '#/components/schemas/ErrorWithCode' patch: operationId: patchPlan tags: [Plans] @@ -394,6 +394,8 @@ paths: $ref: '#/components/responses/Forbidden' '404': $ref: '#/components/responses/NotFound' + '409': + $ref: '#/components/responses/EditorReconciliationRequired' delete: operationId: deletePlan tags: [Plans] @@ -413,6 +415,12 @@ paths: $ref: '#/components/responses/Forbidden' '404': $ref: '#/components/responses/NotFound' + '409': + description: The plan is managed by the schedule editor and still has active or unsettled purchases. Cancel or revoke them first (code managed_plan_has_live_executions). + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorWithCode' /api/plans/{id}/purchases: parameters: @@ -449,11 +457,12 @@ paths: The plan is disabled, has unsupported scheduling settings, no target accounts or insufficient remaining steps, or the requested steps are occupied. Automatic plans also require a valid commitment cap and - the global automatic purchasing switch. + the global automatic purchasing switch. A plan managed by the + schedule editor is refused whole (code editor_reconciliation_required). content: application/json: schema: - $ref: '#/components/schemas/Error' + $ref: '#/components/schemas/ErrorWithCode' # ---- Purchase Actions --------------------------------------------------- /api/purchases/execute: @@ -911,6 +920,8 @@ paths: $ref: '#/components/responses/Forbidden' '404': $ref: '#/components/responses/NotFound' + '409': + $ref: '#/components/responses/EditorReconciliationRequired' /api/purchases/planned/{id}/run: parameters: @@ -2348,6 +2359,14 @@ components: application/json: schema: $ref: '#/components/schemas/Error' + EditorReconciliationRequired: + description: > + The plan is managed by the schedule editor and this legacy change + needs the editor (code editor_reconciliation_required). + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorWithCode' InternalError: description: Internal server error content: @@ -2425,6 +2444,16 @@ components: error: type: string + ErrorWithCode: + type: object + required: [error] + properties: + error: + type: string + code: + type: string + description: Machine-readable reason, for example editor_reconciliation_required. + StatusResponse: type: object required: [status] diff --git a/internal/api/openapi_409_code_test.go b/internal/api/openapi_409_code_test.go new file mode 100644 index 00000000..8ddcf7b9 --- /dev/null +++ b/internal/api/openapi_409_code_test.go @@ -0,0 +1,58 @@ +package api + +import ( + "os" + "path/filepath" + "runtime" + "testing" + + "github.com/stretchr/testify/require" + "gopkg.in/yaml.v3" +) + +// TestOpenAPI409NamesGuardCode keeps the contract honest: every legacy +// operation that can answer a managed-plan refusal documents a 409 whose +// schema carries the machine-readable code. +func TestOpenAPI409NamesGuardCode(t *testing.T) { + _, thisFile, _, _ := runtime.Caller(0) + data, err := os.ReadFile(filepath.Join(filepath.Dir(thisFile), "openapi.yaml")) + require.NoError(t, err) + var doc struct { + Paths map[string]map[string]yaml.Node `yaml:"paths"` + Components struct { + Responses map[string]struct { + Content map[string]struct { + Schema struct { + Ref string `yaml:"$ref"` + } `yaml:"schema"` + } `yaml:"content"` + } `yaml:"responses"` + } `yaml:"components"` + } + require.NoError(t, yaml.Unmarshal(data, &doc)) + + for _, op := range []struct{ path, method string }{ + {"/api/plans/{id}", "put"}, {"/api/plans/{id}", "patch"}, {"/api/plans/{id}", "delete"}, + {"/api/plans/{id}/purchases", "post"}, {"/api/purchases/planned/{id}/resume", "post"}, + } { + node := doc.Paths[op.path][op.method] + var operation struct { + Responses map[string]struct { + Ref string `yaml:"$ref"` + Content map[string]struct { + Schema struct { + Ref string `yaml:"$ref"` + } `yaml:"schema"` + } `yaml:"content"` + } `yaml:"responses"` + } + require.NoError(t, node.Decode(&operation), "%s %s", op.method, op.path) + r, ok := operation.Responses["409"] + require.True(t, ok, "%s %s must document 409", op.method, op.path) + schema := r.Content["application/json"].Schema.Ref + if r.Ref != "" { + schema = doc.Components.Responses[filepath.Base(r.Ref)].Content["application/json"].Schema.Ref + } + require.Equal(t, "#/components/schemas/ErrorWithCode", schema, "%s %s", op.method, op.path) + } +} diff --git a/internal/api/plan_editor_errors.go b/internal/api/plan_editor_errors.go index 290b4e0b..edffb30c 100644 --- a/internal/api/plan_editor_errors.go +++ b/internal/api/plan_editor_errors.go @@ -6,6 +6,10 @@ import "net/http" // legacy write refused because the plan is owned by the schedule editor. const codeEditorReconciliationRequired = "editor_reconciliation_required" +// codeManagedPlanLiveExecutions is the body code for deleting a managed plan +// that still has active or unsettled purchases. +const codeManagedPlanLiveExecutions = "managed_plan_has_live_executions" + // planEditorRefusal builds a client error carrying a machine-readable code. func planEditorRefusal(status int, code, msg string) error { return NewClientErrorWithDetails(status, msg, map[string]any{"code": code}) diff --git a/internal/config/errors.go b/internal/config/errors.go index ac8f63e2..75982d6a 100644 --- a/internal/config/errors.go +++ b/internal/config/errors.go @@ -20,6 +20,10 @@ var ErrPurchasePlanConflict = errors.New("purchase plan changed") // request: callers map it to 409; a background caller would skip, not retry. var ErrEditorReconciliationRequired = errors.New("plan is managed by the schedule editor") +// ErrManagedPlanLiveExecutions means deleting an editor-managed plan would +// detach an execution that may still hold or buy capacity. +var ErrManagedPlanLiveExecutions = errors.New("managed plan still has active or unsettled purchases") + // ErrExecutionNotInExpectedStatus is returned by TransitionExecutionStatus // when the target execution exists but its current status is not in the // allowed `fromStatuses` set -- i.e. the atomic CAS rejected because some diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 3448d9d8..e7aaea8b 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -867,18 +867,24 @@ func (s *PostgresStore) StampPlanNotificationSent(ctx context.Context, planID st // DeletePurchasePlan deletes a purchase plan. func (s *PostgresStore) DeletePurchasePlan(ctx context.Context, planID string) error { - query := `DELETE FROM purchase_plans WHERE id = $1` - - result, err := s.db.Exec(ctx, query, planID) - if err != nil { - return fmt.Errorf("failed to delete purchase plan: %w", err) - } - - if result.RowsAffected() == 0 { - return fmt.Errorf("purchase plan not found: %s", planID) - } - - return nil + return s.WithTx(ctx, func(tx pgx.Tx) error { + plan, err := s.LockPurchasePlanTx(ctx, tx, planID) + if err != nil { + return err + } + if plan == nil { + return fmt.Errorf("purchase plan not found: %s", planID) + } + if plan.RampSchedule.DatesManaged { + if err := requireManagedDeleteTx(ctx, tx, planID); err != nil { + return err + } + } + if _, err := tx.Exec(ctx, `DELETE FROM purchase_plans WHERE id = $1`, planID); err != nil { + return fmt.Errorf("failed to delete purchase plan: %w", err) + } + return nil + }) } // buildListPlansQuery returns the SQL query and args for ListPurchasePlans. @@ -1237,13 +1243,35 @@ func (s *PostgresStore) TransitionExecutionStatusTokenPinned(ctx context.Context // RescheduleAwaitingExecution is a conditional single-column update: the // status predicate makes a lost race a benign no-op instead of an overwrite. func (s *PostgresStore) RescheduleAwaitingExecution(ctx context.Context, executionID string, scheduledDate time.Time) (bool, error) { - tag, err := s.db.Exec(ctx, ` - UPDATE purchase_executions SET scheduled_date = $2, updated_at = NOW() - WHERE execution_id = $1 AND status IN ('pending', 'notified')`, executionID, scheduledDate) + var planID *string + err := s.db.QueryRow(ctx, `SELECT plan_id::text FROM purchase_executions WHERE execution_id = $1`, executionID).Scan(&planID) + if errors.Is(err, pgx.ErrNoRows) { + return false, nil + } if err != nil { return false, fmt.Errorf("reschedule execution %s: %w", executionID, err) } - return tag.RowsAffected() == 1, nil + updated := false + err = s.WithTx(ctx, func(tx pgx.Tx) error { + if planID != nil { + plan, lockErr := s.LockPurchasePlanTx(ctx, tx, *planID) + if lockErr != nil { + return lockErr + } + if plan != nil && plan.RampSchedule.DatesManaged { + return fmt.Errorf("%w: reschedule of execution %s", ErrEditorReconciliationRequired, executionID) + } + } + tag, execErr := tx.Exec(ctx, ` + UPDATE purchase_executions SET scheduled_date = $2, updated_at = NOW() + WHERE execution_id = $1 AND status IN ('pending', 'notified')`, executionID, scheduledDate) + updated = execErr == nil && tag.RowsAffected() == 1 + return execErr + }) + if err != nil { + return false, fmt.Errorf("reschedule execution %s: %w", executionID, err) + } + return updated, nil } func (s *PostgresStore) transitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmpty *bool, tokenHash *string) (*PurchaseExecution, error) { diff --git a/internal/config/store_postgres_managed_guard.go b/internal/config/store_postgres_managed_guard.go index 9ca35d9c..fc4b0064 100644 --- a/internal/config/store_postgres_managed_guard.go +++ b/internal/config/store_postgres_managed_guard.go @@ -118,3 +118,26 @@ func sameAccountSet(a, b []string) bool { } return true } + +// requireManagedDeleteTx refuses deleting a managed plan while an execution may +// hold or buy capacity: the FK would detach it (ON DELETE SET NULL) and it +// would then buy with no plan. Judged under the caller's plan lock, so a root +// inserted by a command that committed first is seen. Settled history +// (completed, canceled, expired, failed without submission) does not block. +func requireManagedDeleteTx(ctx context.Context, tx pgx.Tx, planID string) error { + rows, err := tx.Query(ctx, `SELECT status, no_submission FROM purchase_executions WHERE plan_id = $1`, planID) + if err != nil { + return fmt.Errorf("failed to read plan executions: %w", err) + } + defer rows.Close() + for rows.Next() { + var e PurchaseExecution + if err := rows.Scan(&e.Status, &e.NoSubmission); err != nil { + return fmt.Errorf("failed to scan plan execution: %w", err) + } + if holdsCapacity(&e) && e.Status != StatusCompleted && e.Status != StatusPartiallyCompleted { + return fmt.Errorf("%w: plan %s", ErrManagedPlanLiveExecutions, planID) + } + } + return rows.Err() +} diff --git a/internal/config/store_postgres_pgxmock_test.go b/internal/config/store_postgres_pgxmock_test.go index 37b18d1a..a67dd202 100644 --- a/internal/config/store_postgres_pgxmock_test.go +++ b/internal/config/store_postgres_pgxmock_test.go @@ -630,13 +630,28 @@ func TestPGXMock_UpdatePurchasePlan_NotFound(t *testing.T) { // ─── DeletePurchasePlan ─────────────────────────────────────────────────────── +func expectLockedUnmanagedPlan(mock pgxmock.PgxPoolIface) { + now := time.Now().Truncate(time.Second) + svcJSON, _ := json.Marshal(map[string]ServiceConfig{}) + rampJSON, _ := json.Marshal(RampSchedule{}) + mock.ExpectBegin() + mock.ExpectQuery("FOR UPDATE").WithArgs("plan-id").WillReturnRows(pgxmock.NewRows([]string{ + "id", "name", "enabled", "auto_purchase", "notification_days_before", + "services", "ramp_schedule", "created_at", "updated_at", + "next_execution_date", "last_execution_date", "last_notification_sent", "max_step_commitment", + }).AddRow("plan-id", "My Plan", true, false, 3, svcJSON, rampJSON, now, now, + sql.NullTime{}, sql.NullTime{}, sql.NullTime{}, (*float64)(nil))) +} + func TestPGXMock_DeletePurchasePlan_Success(t *testing.T) { mock := newMock(t) store := storeWith(mock) ctx := context.Background() + expectLockedUnmanagedPlan(mock) mock.ExpectExec("DELETE").WithArgs(pgxmock.AnyArg()). WillReturnResult(pgxmock.NewResult("DELETE", 1)) + mock.ExpectCommit() err := store.DeletePurchasePlan(ctx, "plan-id") require.NoError(t, err) @@ -648,8 +663,9 @@ func TestPGXMock_DeletePurchasePlan_NotFound(t *testing.T) { store := storeWith(mock) ctx := context.Background() - mock.ExpectExec("DELETE").WithArgs(pgxmock.AnyArg()). - WillReturnResult(pgxmock.NewResult("DELETE", 0)) + mock.ExpectBegin() + mock.ExpectQuery("FOR UPDATE").WithArgs("plan-id").WillReturnError(pgx.ErrNoRows) + mock.ExpectRollback() err := store.DeletePurchasePlan(ctx, "plan-id") require.Error(t, err)