fix(history): sort future pending ramp steps after due rows in the actionable fetch

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"])
+}

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

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)

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)
+}

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())
+}

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)