From 44d06e5bbf7c984154d6d1f40e24e25554263ffc Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 16 Oct 2026 23:22:32 +0100 Subject: [PATCH] fix(history): sort future pending ramp steps after due rows in the actionable fetch GetExecutionsByStatuses ordered by scheduled_date DESC with a MaxListLimit (1000) clamp. Plan creation writes up to 52 pending ramp steps per plan, all dated in the future, so about 20 plans fill the cap with steps months ahead and a pending approval due today never comes back (truncated was set, but the row was lost). Pending rows with a future scheduled_date now sort after every other row, soonest first among themselves, so the cap drops the farthest steps. Every other row keeps scheduled_date DESC; terminal statuses never match the predicate, so the terminal class is unchanged. The History handler re-sorts the merged list by timestamp, so only which rows survive the cap changes. The shared mock mirrors the new order. Closes #680 Co-Authored-By: Claude Opus 5.5 --- ...dler_history_actionable_starvation_test.go | 34 ++++++ internal/config/interfaces.go | 4 +- internal/config/store_postgres.go | 14 ++- ...executions_by_statuses_integration_test.go | 109 ++++++++++++++++++ .../config/store_postgres_pgxmock_test.go | 22 +++- internal/mocks/stores.go | 21 +++- 6 files changed, 199 insertions(+), 5 deletions(-) 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)