From 67eed5725b8045b3780c1d0f7ca80d6a8bb17f04 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 15 Oct 2026 17:50:23 +0200 Subject: [PATCH] fix(purchase): claim the exact managed schedule root Resolve managed claims and manual due notifications from the locked next root instead of a cached plan date. Stop on missing or occupied roots and recheck outcomes after waiting for execution locks. --- ..._postgres_managed_root_integration_test.go | 29 ++ .../store_postgres_plan_schedule_clock.go | 27 ++ internal/purchase/notifications.go | 5 +- internal/purchase/plan_step_attach.go | 6 + internal/purchase/plan_step_managed.go | 58 ++++ .../plan_step_managed_integration_test.go | 272 ++++++++++++++++++ 6 files changed, 396 insertions(+), 1 deletion(-) create mode 100644 internal/config/store_postgres_managed_root_integration_test.go create mode 100644 internal/purchase/plan_step_managed.go create mode 100644 internal/purchase/plan_step_managed_integration_test.go diff --git a/internal/config/store_postgres_managed_root_integration_test.go b/internal/config/store_postgres_managed_root_integration_test.go new file mode 100644 index 00000000..38262847 --- /dev/null +++ b/internal/config/store_postgres_managed_root_integration_test.go @@ -0,0 +1,29 @@ +//go:build integration + +package config + +import ( + "context" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/require" +) + +func TestManagedRootCapabilityRefusesUnmanaged(t *testing.T) { + ctx := t.Context() + f := newRampFixture(ctx, t, 0, 0, 3) + capability, ok := any(f.store).(interface { + ResolveManagedPlanRootTx(context.Context, pgx.Tx, *PurchasePlan) (*PurchaseExecution, error) + }) + if !ok { + t.Skip("parent lacks the capability; behavioral parent proof uses the actual Manager") + } + require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error { + p, err := f.store.LockPurchasePlanTx(ctx, tx, f.planID) + require.NoError(t, err) + _, err = capability.ResolveManagedPlanRootTx(ctx, tx, p) + require.Error(t, err) + return nil + })) +} diff --git a/internal/config/store_postgres_plan_schedule_clock.go b/internal/config/store_postgres_plan_schedule_clock.go index 820ee1d6..e7ab2336 100644 --- a/internal/config/store_postgres_plan_schedule_clock.go +++ b/internal/config/store_postgres_plan_schedule_clock.go @@ -47,6 +47,33 @@ func (s *PostgresStore) planScheduleNextDateTx(ctx context.Context, tx pgx.Tx, p return &root.ScheduledDate, nil } +// ResolveManagedPlanRootTx requires the plan row lock before execution locks. +func (s *PostgresStore) ResolveManagedPlanRootTx(ctx context.Context, tx pgx.Tx, plan *PurchasePlan) (*PurchaseExecution, error) { + if plan == nil || !plan.RampSchedule.DatesManaged { + return nil, fmt.Errorf("managed root resolution requires a managed plan") + } + if plan.RampSchedule.IsComplete() { + return nil, nil + } + rows, err := tx.Query(ctx, `SELECT execution_id FROM purchase_executions + WHERE plan_id=$1 AND step_number=$2 ORDER BY execution_id FOR UPDATE`, plan.ID, plan.RampSchedule.CurrentStep+1) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, err + } + } + if err := rows.Err(); err != nil { + return nil, err + } + rows.Close() + return s.managedPlanNextRootTx(ctx, tx, plan) +} + func (s *PostgresStore) managedPlanNextRootTx(ctx context.Context, tx pgx.Tx, plan *PurchasePlan) (*PurchaseExecution, error) { rows, err := s.ListExecutionsForPlanStepTx(ctx, tx, plan.ID, plan.RampSchedule.CurrentStep+1) if err != nil { diff --git a/internal/purchase/notifications.go b/internal/purchase/notifications.go index 83a7935f..0cf946ba 100644 --- a/internal/purchase/notifications.go +++ b/internal/purchase/notifications.go @@ -130,9 +130,12 @@ func (m *Manager) dueManualRootTx(ctx context.Context, tx pgx.Tx, planID string) if err != nil { return nil, err } - if locked == nil || !locked.Enabled || locked.AutoPurchase { + if !enabledManualPlan(locked) { return nil, nil } + if locked.RampSchedule.DatesManaged { + return m.dueManagedPlanRootTx(ctx, tx, locked) + } rows, err := m.config.ListExecutionsForPlanStepTx(ctx, tx, planID, locked.RampSchedule.CurrentStep+1) if err != nil || len(rows) == 0 { return nil, err diff --git a/internal/purchase/plan_step_attach.go b/internal/purchase/plan_step_attach.go index 743c5ef2..1008b9e8 100644 --- a/internal/purchase/plan_step_attach.go +++ b/internal/purchase/plan_step_attach.go @@ -66,6 +66,12 @@ func (m *Manager) claimPlanStep(ctx context.Context, tx pgx.Tx, plan *config.Pur if locked == nil { return stepClaim{}, fmt.Errorf("plan not found: %s", plan.ID) } + if locked.RampSchedule.DatesManaged != plan.RampSchedule.DatesManaged { + return stepClaim{}, errPlanStepSkipped + } + if locked.RampSchedule.DatesManaged { + return m.claimManagedPlanStep(ctx, tx, plan, locked, providers) + } if !planUnchanged(plan, locked, providers) { return stepClaim{}, fmt.Errorf("%w: plan %s changed since it was listed", errPlanStepSkipped, plan.ID) } diff --git a/internal/purchase/plan_step_managed.go b/internal/purchase/plan_step_managed.go new file mode 100644 index 00000000..991a49bb --- /dev/null +++ b/internal/purchase/plan_step_managed.go @@ -0,0 +1,58 @@ +package purchase + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/jackc/pgx/v5" +) + +type managedPlanRootStore interface { + ResolveManagedPlanRootTx(context.Context, pgx.Tx, *config.PurchasePlan) (*config.PurchaseExecution, error) +} + +func (m *Manager) managedPlanRootTx(ctx context.Context, tx pgx.Tx, plan *config.PurchasePlan) (*config.PurchaseExecution, error) { + store, ok := m.config.(managedPlanRootStore) + if !ok { + return nil, fmt.Errorf("%w: managed root resolution unavailable", errPlanStepSkipped) + } + return store.ResolveManagedPlanRootTx(ctx, tx, plan) +} + +func (m *Manager) claimManagedPlanStep(ctx context.Context, tx pgx.Tx, listed, locked *config.PurchasePlan, providers []string) (stepClaim, error) { + if !managedPlanUnchanged(listed, locked, providers) { + return stepClaim{}, errPlanStepSkipped + } + root, err := m.managedPlanRootTx(ctx, tx, locked) + if err != nil { + return stepClaim{}, err + } + if root == nil || root.Status == "scheduled" { + return stepClaim{}, errPlanStepSkipped + } + locked.NextExecutionDate = &root.ScheduledDate + return m.claimFromRows(ctx, tx, locked, stepClaim{number: locked.RampSchedule.CurrentStep + 1}, []config.PurchaseExecution{*root}) +} + +func managedPlanUnchanged(listed, locked *config.PurchasePlan, providers []string) bool { + return locked.Enabled && locked.AutoPurchase == listed.AutoPurchase && locked.UpdatedAt.Equal(listed.UpdatedAt) && + locked.RampSchedule.CurrentStep == listed.RampSchedule.CurrentStep && strings.Join(planProviders(locked), ",") == strings.Join(providers, ",") +} + +func enabledManualPlan(plan *config.PurchasePlan) bool { + return plan != nil && plan.Enabled && !plan.AutoPurchase +} + +func (m *Manager) dueManagedPlanRootTx(ctx context.Context, tx pgx.Tx, plan *config.PurchasePlan) (*config.PurchaseExecution, error) { + root, err := m.managedPlanRootTx(ctx, tx, plan) + if err != nil || root == nil { + return nil, err + } + if root.Status != "scheduled" && len(root.Recommendations) == 0 && !root.ScheduledDate.After(time.Now()) { + return root, nil + } + return nil, nil +} diff --git a/internal/purchase/plan_step_managed_integration_test.go b/internal/purchase/plan_step_managed_integration_test.go new file mode 100644 index 00000000..d8824f79 --- /dev/null +++ b/internal/purchase/plan_step_managed_integration_test.go @@ -0,0 +1,272 @@ +//go:build integration + +package purchase + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/require" +) + +// Native store and Manager; recommendation resolver and email are recordings. +type managedClaimFixture struct { + ctx context.Context + store *config.PostgresStore + pool *pgxpool.Pool + plan *config.PurchasePlan + m *Manager + resolver *fakeResolver + sender *countingSender +} + +func newManagedClaimFixture(t *testing.T) *managedClaimFixture { + t.Helper() + ctx := t.Context() + store, pool := rampStepStore(ctx, t) + plan := attachPlanInDB(ctx, t, store, "managed-claim", time.Now().AddDate(0, 0, 47)) + resolver, sender := &fakeResolver{recs: []config.RecommendationRecord{attachRec("fixture", 1)}}, newCountingSender(ctx) + return &managedClaimFixture{ctx, store, pool, plan, &Manager{config: store, email: sender, resolver: resolver}, resolver, sender} +} + +func (f *managedClaimFixture) row(t *testing.T, step int, status string) *config.PurchaseExecution { + t.Helper() + e := &config.PurchaseExecution{PlanID: f.plan.ID, ExecutionID: uuid.NewString(), StepNumber: step, Status: status, ScheduledDate: *f.plan.NextExecutionDate, Recommendations: []config.RecommendationRecord{attachRec("fixture", 1)}} + require.NoError(t, f.store.SavePurchaseExecution(f.ctx, e)) + return e +} + +func (f *managedClaimFixture) seal(t *testing.T) { + t.Helper() + _, err := f.pool.Exec(f.ctx, `UPDATE purchase_plans SET ramp_schedule=ramp_schedule||'{"dates_managed":true}' WHERE id=$1`, f.plan.ID) + require.NoError(t, err) + f.plan, err = f.store.GetPurchasePlan(f.ctx, f.plan.ID) + require.NoError(t, err) +} + +func TestManagedClaimConservativeRoots(t *testing.T) { + for _, tc := range []struct { + name, status, sql, extra string + adopt bool + }{ + {name: "missing", extra: "later"}, + {name: "canceled", status: "canceled"}, {name: "failed", status: "failed"}, {name: "paused", status: "paused"}, {name: "uncertain", status: "approved"}, + {name: "duplicate", status: "pending", extra: "duplicate"}, + {name: "child", status: "pending", extra: "child"}, + {name: "retry", status: "pending", sql: "retry_attempt_n=1"}, + {name: "lineage", status: "pending", sql: "idempotency_key='fixture:child'"}, + {name: "outgoing", status: "pending", extra: "outgoing"}, + {name: "canceled acted", status: "pending", extra: "acted"}, + {name: "executed", status: "pending", sql: "executed_at=NOW()"}, + {name: "completed", status: "pending", sql: "completed_at=NOW()"}, + {name: "purchased", status: "pending", sql: "recommendations=jsonb_set(recommendations,'{0,purchased}','true')"}, + {name: "skip", status: "pending", sql: "pre_approval_skip_reason='prior'"}, + {name: "zero date", status: "pending", sql: "scheduled_date='0001-01-01T00:00:00Z'"}, + {name: "pending timer", status: "pending", sql: "scheduled_execution_at=NOW()"}, + {name: "scheduled missing timer", status: "scheduled"}, + {name: "scheduled", status: "scheduled", sql: "scheduled_execution_at=NOW()"}, + {name: "incoming step", status: "pending", extra: "incoming"}, + {name: "incoming foreign", status: "pending", extra: "foreign"}, + {name: "pending", status: "pending", adopt: true}, {name: "notified", status: "notified", adopt: true}, + {name: "replacement", status: "pending", extra: "unacted", adopt: true}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newManagedClaimFixture(t) + var root *config.PurchaseExecution + if tc.status != "" { + root = f.row(t, 1, tc.status) + } + if tc.sql != "" { + _, err := f.pool.Exec(f.ctx, "UPDATE purchase_executions SET "+tc.sql+" WHERE execution_id=$1", root.ExecutionID) + require.NoError(t, err) + } + switch tc.extra { + case "later": + f.row(t, 2, "pending") + case "duplicate": + f.row(t, 1, "pending") + case "acted", "unacted": + old := f.row(t, 1, "canceled") + if tc.extra == "acted" { + _, err := f.pool.Exec(f.ctx, `UPDATE purchase_executions SET completed_at=NOW() WHERE execution_id=$1`, old.ExecutionID) + require.NoError(t, err) + } + case "child": + var account string + require.NoError(t, f.pool.QueryRow(f.ctx, `INSERT INTO cloud_accounts(name,provider,external_id) VALUES('managed-claim','aws','123456789012') RETURNING id`).Scan(&account)) + _, err := f.pool.Exec(f.ctx, `UPDATE purchase_executions SET cloud_account_id=$2 WHERE execution_id=$1`, root.ExecutionID, account) + require.NoError(t, err) + case "outgoing", "incoming", "foreign": + other := f.row(t, 2, "failed") + if tc.extra == "foreign" { + p := attachPlanInDB(f.ctx, t, f.store, "foreign", *f.plan.NextExecutionDate) + _, err := f.pool.Exec(f.ctx, `UPDATE purchase_executions SET plan_id=$2 WHERE execution_id=$1`, other.ExecutionID, p.ID) + require.NoError(t, err) + } + from, to := other.ExecutionID, root.ExecutionID + if tc.extra == "outgoing" { + from, to = to, from + } + _, err := f.pool.Exec(f.ctx, `UPDATE purchase_executions SET retry_execution_id=$2 WHERE execution_id=$1`, from, to) + require.NoError(t, err) + } + f.seal(t) + before := stepRowsInDB(f.ctx, t, f.store, f.plan.ID) + exec, token, _, err := f.m.getOrCreateExecution(f.ctx, f.plan) + if tc.adopt { + require.NoError(t, err) + require.Equal(t, root.ExecutionID, exec.ExecutionID) + require.NotEmpty(t, token) + require.True(t, root.ScheduledDate.Equal(exec.ScheduledDate)) + } else { + require.Error(t, err, "managed claim must stop") + require.Nil(t, exec) + require.Empty(t, token) + } + require.Equal(t, before, stepRowsInDB(f.ctx, t, f.store, f.plan.ID)) + require.Zero(t, f.resolver.calls.Load()) + require.Zero(t, f.sender.sends) + }) + } +} + +func TestManagedClaimEligibilityAndClock(t *testing.T) { + for _, name := range []string{"disabled", "finished", "provider", "auto", "version", "step", "marker", "nil clock", "past clock", "empty", "inside veto"} { + t.Run(name, func(t *testing.T) { + f := newManagedClaimFixture(t) + root := f.row(t, 1, "pending") + if name == "disabled" || name == "finished" { + _, err := f.pool.Exec(f.ctx, `UPDATE purchase_plans SET enabled=$2,ramp_schedule=jsonb_set(ramp_schedule,'{total_steps}',$3::jsonb) WHERE id=$1`, f.plan.ID, name != "disabled", map[bool]string{true: "0", false: "4"}[name == "finished"]) + require.NoError(t, err) + } + if name == "nil clock" || name == "past clock" { + var clock *time.Time + if name == "past clock" { + date := time.Now().Add(-time.Hour) + clock = &date + } + _, err := f.pool.Exec(f.ctx, `UPDATE purchase_plans SET next_execution_date=$2 WHERE id=$1`, f.plan.ID, clock) + require.NoError(t, err) + } + if name == "empty" || name == "inside veto" { + _, err := f.pool.Exec(f.ctx, `UPDATE purchase_executions SET recommendations='[]',scheduled_date=CASE WHEN $2 THEN NOW()+INTERVAL '1 day' ELSE scheduled_date END WHERE execution_id=$1`, root.ExecutionID, name == "inside veto") + require.NoError(t, err) + } + f.seal(t) + switch name { + case "provider": + svc := f.plan.Services["ec2"] + svc.Provider = "azure" + f.plan.Services["ec2"] = svc + case "auto": + f.plan.AutoPurchase = false + case "version": + f.plan.UpdatedAt = f.plan.UpdatedAt.Add(-time.Second) + case "step": + f.plan.RampSchedule.CurrentStep++ + case "marker": + f.plan.RampSchedule.DatesManaged = false + } + before := stepRowsInDB(f.ctx, t, f.store, f.plan.ID) + exec, token, _, err := f.m.getOrCreateExecution(f.ctx, f.plan) + if name == "nil clock" || name == "past clock" || name == "empty" { + require.NoError(t, err) + require.Equal(t, root.ExecutionID, exec.ExecutionID) + require.True(t, root.ScheduledDate.Equal(exec.ScheduledDate)) + require.NotEmpty(t, token) + } else { + require.Error(t, err) + require.Nil(t, exec) + require.Empty(t, token) + } + require.Len(t, stepRowsInDB(f.ctx, t, f.store, f.plan.ID), 1) + if name == "inside veto" { + require.Equal(t, before, stepRowsInDB(f.ctx, t, f.store, f.plan.ID)) + } + require.EqualValues(t, map[bool]int{true: 1, false: 0}[name == "empty"], f.resolver.calls.Load()) + }) + } +} + +func TestManagedManualDue(t *testing.T) { + for _, state := range []string{"pending", "scheduled", "finished", "missing", "occupied"} { + f := newManagedClaimFixture(t) + root := f.row(t, 1, "pending") + _, err := f.pool.Exec(f.ctx, `UPDATE purchase_executions SET recommendations=CASE WHEN $4 THEN recommendations ELSE '[]'::jsonb END,step_number=CASE WHEN $3 THEN 2 ELSE step_number END,scheduled_date=NOW()-INTERVAL '1 day',status=$2::text,scheduled_execution_at=CASE WHEN $2::text='scheduled' THEN NOW() END WHERE execution_id=$1`, root.ExecutionID, map[bool]string{true: "scheduled", false: "pending"}[state == "scheduled"], state == "missing", state == "occupied") + require.NoError(t, err) + _, err = f.pool.Exec(f.ctx, `UPDATE purchase_plans SET auto_purchase=false,ramp_schedule=jsonb_set(ramp_schedule,'{total_steps}',$2::jsonb) WHERE id=$1`, f.plan.ID, map[bool]string{true: "0", false: "4"}[state == "finished"]) + require.NoError(t, err) + f.seal(t) + var got *config.PurchaseExecution + require.NoError(t, f.store.WithTx(f.ctx, func(tx pgx.Tx) error { + var readErr error + got, readErr = f.m.dueManualRootTx(f.ctx, tx, f.plan.ID) + return readErr + })) + if state == "pending" { + require.NotNil(t, got) + require.Equal(t, root.ExecutionID, got.ExecutionID) + } else { + require.Nil(t, got) + } + } +} + +func TestManagedClaimWaitsForOutcome(t *testing.T) { + f := newManagedClaimFixture(t) + root := f.row(t, 1, "pending") + f.seal(t) + ctx, cancel := context.WithTimeout(f.ctx, 20*time.Second) + defer cancel() + holder, err := f.pool.Begin(ctx) + require.NoError(t, err) + defer func() { _ = holder.Rollback(context.Background()) }() + _, err = holder.Exec(ctx, `SELECT execution_id FROM purchase_executions WHERE execution_id=$1 FOR UPDATE`, root.ExecutionID) + require.NoError(t, err) + type claimResult struct { + token string + err error + } + done := make(chan claimResult, 1) + go func() { + _, token, _, claimErr := f.m.getOrCreateExecution(ctx, f.plan) + done <- claimResult{token, claimErr} + }() + joined := false + defer func() { + cancel() + if !joined { + select { + case <-done: + case <-time.After(3 * time.Second): + t.Error("claim did not join") + } + } + }() + require.Eventually(t, func() bool { + var n int + readErr := f.pool.QueryRow(ctx, `SELECT count(*) FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock' AND query LIKE 'SELECT execution_id FROM purchase_executions%FOR UPDATE%'`).Scan(&n) + return readErr == nil && n > 0 + }, 5*time.Second, 20*time.Millisecond) + root.Status = "paused" + require.NoError(t, f.store.SaveExecutionIfStatusTx(ctx, holder, root, []string{"pending"})) + require.NoError(t, holder.Commit(ctx)) + select { + case result := <-done: + joined = true + require.Error(t, result.err) + require.Empty(t, result.token) + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + require.Zero(t, f.resolver.calls.Load()) + rows := stepRowsInDB(f.ctx, t, f.store, f.plan.ID) + require.Len(t, rows, 1) + require.Equal(t, root.ExecutionID, rows[0].ExecutionID) + require.Equal(t, "paused", rows[0].Status) +}