diff --git a/internal/api/handler_history_actionable_starvation_test.go b/internal/api/handler_history_actionable_starvation_test.go index 15efcfbb0..e3aa16612 100644 --- a/internal/api/handler_history_actionable_starvation_test.go +++ b/internal/api/handler_history_actionable_starvation_test.go @@ -169,3 +169,37 @@ func TestHandler_getHistory_StaleSweepBoundedToDefaultListLimit(t *testing.T) { store.AssertNumberOfCalls(t, "ExpireExecutionAtomic", config.DefaultListLimit) } + +// TestHandler_getHistory_DueApprovalNotCrowdedOutByFutureRampSteps is the +// issue #680 regression through the real getHistory path: 20 plans x 52 +// future pending ramp steps exceed the actionable cap, and the pending +// approval due today must still be returned, with truncated set. +func TestHandler_getHistory_DueApprovalNotCrowdedOutByFutureRampSteps(t *testing.T) { + ctx := context.Background() + now := time.Now() + var execs []config.PurchaseExecution + for p := 0; p < 20; p++ { + for s := 1; s <= 52; s++ { + execs = append(execs, config.PurchaseExecution{ + ExecutionID: fmt.Sprintf("ramp-%d-%d", p, s), + Status: "pending", + ScheduledDate: now.AddDate(0, 0, 7*s), + }) + } + } + execs = append(execs, executionsWithStatus(1, "pending", now.Add(-time.Hour), "due")...) + store := newStarvationStore(ctx) + store.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(execs, nil) + + body := runStarvationHistory(t, ctx, store) + + ids := map[string]bool{} + for _, p := range body["purchases"].([]any) { + ids[p.(map[string]any)["purchase_id"].(string)] = true + } + assert.True(t, ids["due-0"], "pending approval due today must survive 1040 future ramp steps") + assert.True(t, ids["ramp-0-1"], "the soonest future ramp step survives the cap") + assert.False(t, ids["ramp-0-52"], "the farthest future ramp steps are the ones dropped") + assert.Len(t, body["purchases"], config.MaxListLimit) + assert.Equal(t, true, body["truncated"]) +} diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index b8ac4fe5c..666d3b14d 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -147,7 +147,9 @@ type StoreInterface interface { GetExecutionFanOutShape(ctx context.Context, lineageKey string) (string, error) GetPendingExecutions(ctx context.Context) ([]PurchaseExecution, error) // GetExecutionsByStatuses returns executions in any of the given states, - // newest first, capped at `limit`. Used by the History handler to render + // newest first, capped at `limit`, except that pending rows dated in the + // future (not-yet-due ramp steps) sort last, soonest first, so they cannot + // push a due row past the cap (issue #680). Used by the History handler to render // pending + failed + expired alongside completed purchases; the scheduler // keeps using GetPendingExecutions (which is narrower and doesn't share // this method's status filter) to avoid accidental double-processing of diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index e7aaea8bd..4a852a8ae 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -1481,6 +1481,16 @@ func (s *PostgresStore) CancelScheduledExecutionAtomic(ctx context.Context, tx p // without changing the narrower GetPendingExecutions contract the scheduler // depends on. // +// Pending rows dated in the future are the one exception to newest-first +// (issue #680): plan creation writes every ramp step up front as a pending +// row, up to 52 per plan, so about 20 plans fill MaxListLimit with steps +// months ahead and a plain scheduled_date DESC would cut off a pending +// approval due today. Those future pending steps sort after every other row, +// and among themselves soonest first, so the cap drops the farthest steps. +// No other status is affected: a notified, scheduled or in-flight row is +// actionable whatever its date, and terminal statuses never match the +// pending predicate, so their order is unchanged. +// // Executions with status 'completed' and an empty error are excluded: they are // already represented by their purchase_history rows, so returning them would // only consume the cap (the History handler skips them anyway). @@ -1506,7 +1516,9 @@ func (s *PostgresStore) GetExecutionsByStatuses(ctx context.Context, statuses [] FROM purchase_executions WHERE status = ANY($1) AND NOT (status = 'completed' AND COALESCE(error, '') = '') - ORDER BY scheduled_date DESC + ORDER BY (status = 'pending' AND scheduled_date > now()) ASC, + CASE WHEN status = 'pending' AND scheduled_date > now() THEN scheduled_date END ASC, + scheduled_date DESC LIMIT $2 ` return s.queryExecutions(ctx, query, statuses, limit) diff --git a/internal/config/store_postgres_executions_by_statuses_integration_test.go b/internal/config/store_postgres_executions_by_statuses_integration_test.go index 9de546ba7..7c8764899 100644 --- a/internal/config/store_postgres_executions_by_statuses_integration_test.go +++ b/internal/config/store_postgres_executions_by_statuses_integration_test.go @@ -5,6 +5,7 @@ package config import ( "context" + "fmt" "testing" "time" @@ -54,3 +55,111 @@ func TestIntegration_GetExecutionsByStatuses_RowClasses(t *testing.T) { assert.Equal(t, pending, limited[0].ExecutionID) assert.Equal(t, failed, limited[1].ExecutionID) } + +// TestIntegration_GetExecutionsByStatuses_DueRowSurvivesFutureRampSteps is the +// issue #680 regression. Plan creation writes up to 52 pending ramp steps per +// plan, all dated in the future, so 20 plans put 1040 future pending rows in +// the actionable class. Under a plain scheduled_date DESC those fill the +// MaxListLimit cap and the pending approval due today never comes back. Every +// row that is not a future pending step must come first, newest first as +// before; future pending steps follow, soonest first. +func TestIntegration_GetExecutionsByStatuses_DueRowSurvivesFutureRampSteps(t *testing.T) { + ctx := context.Background() + store := setupRampStepStore(ctx, t) + + now := time.Now() + save := func(planID, status string, scheduled time.Time) string { + exec := &PurchaseExecution{ + PlanID: planID, + ExecutionID: uuid.New().String(), + IdempotencyKey: uuid.New().String(), + Status: status, + ScheduledDate: scheduled, + } + require.NoError(t, store.SavePurchaseExecution(ctx, exec)) + return exec.ExecutionID + } + + const plans, stepsPerPlan = 20, 52 + var planIDs []string + for p := 0; p < plans; p++ { + plan := &PurchasePlan{ + Name: fmt.Sprintf("Ramp %d %s", p, uuid.New().String()[:8]), + Services: map[string]ServiceConfig{}, + RampSchedule: RampSchedule{ + Type: "weekly", PercentPerStep: 2, StepIntervalDays: 7, + TotalSteps: stepsPerPlan, StartDate: now, + }, + } + require.NoError(t, store.CreatePurchasePlan(ctx, plan)) + planIDs = append(planIDs, plan.ID) + for s := 1; s <= stepsPerPlan; s++ { + save(plan.ID, "pending", now.AddDate(0, 0, 7*s).Add(time.Duration(p)*time.Minute)) + } + } + dueToday := save(planIDs[0], "pending", now.Add(-time.Hour)) + overdue := save(planIDs[1], "pending", now.Add(-48*time.Hour)) + // A notified row stays actionable with a future scheduled_date: its + // approval link is already out. + notifiedFuture := save(planIDs[2], "notified", now.Add(72*time.Hour)) + scheduledSoon := save(planIDs[3], "scheduled", now.Add(10*time.Minute)) + + actionable := []string{"pending", "notified", "scheduled", "approved", "running", "paused"} + got, err := store.GetExecutionsByStatuses(ctx, actionable, MaxListLimit) + require.NoError(t, err) + require.Len(t, got, MaxListLimit, "1044 rows still hit the cap, so the History handler sets truncated") + + head := make([]string, 4) + for i := range head { + head[i] = got[i].ExecutionID + } + assert.Equal(t, []string{notifiedFuture, scheduledSoon, dueToday, overdue}, head, + "rows actionable now come first, newest first as before") + + for i := 4; i < len(got); i++ { + require.Equal(t, "pending", got[i].Status) + require.True(t, got[i].ScheduledDate.After(now), "only future pending steps follow") + if i > 4 { + require.False(t, got[i].ScheduledDate.Before(got[i-1].ScheduledDate), + "future pending steps are soonest first, so the next ramp steps survive the cap") + } + } + + small, err := store.GetExecutionsByStatuses(ctx, actionable, 3) + require.NoError(t, err) + require.Len(t, small, 3) + assert.Equal(t, []string{notifiedFuture, scheduledSoon, dueToday}, + []string{small[0].ExecutionID, small[1].ExecutionID, small[2].ExecutionID}) +} + +// TestIntegration_GetExecutionsByStatuses_TerminalOrderUnchanged pins that the +// #680 ordering only moves future pending rows: a canceled ramp step keeps its +// future scheduled_date, and the terminal class must still list newest first. +func TestIntegration_GetExecutionsByStatuses_TerminalOrderUnchanged(t *testing.T) { + ctx := context.Background() + store := setupRampStepStore(ctx, t) + + now := time.Now() + save := func(status string, scheduled time.Time) string { + exec := &PurchaseExecution{ + ExecutionID: uuid.New().String(), + IdempotencyKey: uuid.New().String(), + Status: status, + Error: "x", + ScheduledDate: scheduled, + } + require.NoError(t, store.SavePurchaseExecution(ctx, exec)) + return exec.ExecutionID + } + canceledFuture := save(StatusCanceled, now.AddDate(0, 3, 0)) + failedPast := save("failed", now.Add(-time.Hour)) + expiredOlder := save("expired", now.AddDate(0, 0, -10)) + + got, err := store.GetExecutionsByStatuses(ctx, []string{"failed", "expired", StatusCanceled}, 100) + require.NoError(t, err) + ids := make([]string, len(got)) + for i := range got { + ids[i] = got[i].ExecutionID + } + assert.Equal(t, []string{canceledFuture, failedPast, expiredOlder}, ids) +} diff --git a/internal/config/store_postgres_pgxmock_test.go b/internal/config/store_postgres_pgxmock_test.go index a67dd2029..36e5c8877 100644 --- a/internal/config/store_postgres_pgxmock_test.go +++ b/internal/config/store_postgres_pgxmock_test.go @@ -3248,7 +3248,7 @@ func TestPGXMock_GetExecutionsByStatuses_ExcludesCleanCompleted(t *testing.T) { mock := newMock(t) store := storeWith(mock) - mock.ExpectQuery(`(?s)FROM purchase_executions\s+WHERE status = ANY\(\$1\)\s+AND NOT \(status = 'completed' AND COALESCE\(error, ''\) = ''\)\s+ORDER BY scheduled_date DESC\s+LIMIT \$2`). + mock.ExpectQuery(`(?s)FROM purchase_executions\s+WHERE status = ANY\(\$1\)\s+AND NOT \(status = 'completed' AND COALESCE\(error, ''\) = ''\)\s+ORDER BY .*LIMIT \$2`). WithArgs(pgxmock.AnyArg(), pgxmock.AnyArg()). WillReturnRows(pgxmock.NewRows(stuckExecCols())) @@ -3359,3 +3359,23 @@ func TestPGXMock_UpdateGlobalConfigAtomic_PreservesAutoPurchaseStamps(t *testing require.NoError(t, err) require.NoError(t, mock.ExpectationsWereMet()) } + +// TestPGXMock_GetExecutionsByStatuses_FuturePendingStepsSortLast pins the +// issue #680 ORDER BY: future pending ramp steps sort after every other row +// (soonest first among themselves), and everything else keeps scheduled_date +// DESC. Under a plain scheduled_date DESC, 20 plans of 52 future steps fill +// MaxListLimit and a pending approval due today never comes back. +func TestPGXMock_GetExecutionsByStatuses_FuturePendingStepsSortLast(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + + mock.ExpectQuery(`(?s)ORDER BY \(status = 'pending' AND scheduled_date > now\(\)\) ASC,\s+`+ + `CASE WHEN status = 'pending' AND scheduled_date > now\(\) THEN scheduled_date END ASC,\s+`+ + `scheduled_date DESC\s+LIMIT \$2`). + WithArgs(pgxmock.AnyArg(), MaxListLimit). + WillReturnRows(pgxmock.NewRows(stuckExecCols())) + + _, err := store.GetExecutionsByStatuses(context.Background(), []string{"pending", "notified"}, MaxListLimit) + require.NoError(t, err) + assert.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index e2fbd3e5f..6da82d228 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1820,7 +1820,8 @@ func (m *MockConfigStore) GetExecutionsByStatuses(ctx context.Context, statuses } // filterExecutionsLikeStore applies the store contract to canned rows: filter by -// status, drop clean completed rows, order by ScheduledDate DESC, then cap. +// status, drop clean completed rows, order by ScheduledDate DESC with future +// pending rows last and soonest first (issue #680), then cap. func filterExecutionsLikeStore(rows []config.PurchaseExecution, statuses []string, limit int) []config.PurchaseExecution { if len(statuses) == 0 { return nil @@ -1837,13 +1838,29 @@ func filterExecutionsLikeStore(rows []config.PurchaseExecution, statuses []strin } out = append(out, *r) } - sort.SliceStable(out, func(i, j int) bool { return out[i].ScheduledDate.After(out[j].ScheduledDate) }) + now := time.Now() + sort.SliceStable(out, func(i, j int) bool { return executionSortsBefore(&out[i], &out[j], now) }) if len(out) > limit { out = out[:limit] } return out } +// executionSortsBefore mirrors GetExecutionsByStatuses' ORDER BY: rows that +// are not future pending steps first, newest first; future pending steps +// after them, soonest first. +func executionSortsBefore(a, b *config.PurchaseExecution, now time.Time) bool { + aFuture := a.Status == "pending" && a.ScheduledDate.After(now) + bFuture := b.Status == "pending" && b.ScheduledDate.After(now) + if aFuture != bFuture { + return bFuture + } + if aFuture { + return a.ScheduledDate.Before(b.ScheduledDate) + } + return a.ScheduledDate.After(b.ScheduledDate) +} + // CountExecutionsByPlanAndStatus mocks the CountExecutionsByPlanAndStatus operation. func (m *MockConfigStore) CountExecutionsByPlanAndStatus(ctx context.Context, statuses []string, since time.Time) (map[string]config.ExecutionStatusCounts, error) { m.record("CountExecutionsByPlanAndStatus", ctx, statuses, since)