fix(purchase): skip scheduled purchases due more than 24h ago in the fire and process sweeps

.env.example

@@ -123,4 +123,5 @@ DASHBOARD_URL=http://localhost:3000
 # CUDLY_MAX_ACCOUNT_PARALLELISM=8
 # DEFAULT_PAYMENT_OPTION=no-upfront
 # DEFAULT_RAMP_SCHEDULE=quarterly
+# SCHEDULED_PURCHASE_MAX_OVERDUE=24h
 # ENVIRONMENT=local

go.mod

@@ -59,7 +59,7 @@ require (
 	go.opentelemetry.io/otel/metric v1.45.0 // indirect
 	go.opentelemetry.io/otel/trace v1.45.0 // indirect
 	golang.org/x/crypto v0.58.0
-	golang.org/x/net v0.61.0 // indirect
+	golang.org/x/net v0.62.0 // indirect
 	golang.org/x/oauth2 v0.37.0
 	golang.org/x/sync v0.24.0
 	golang.org/x/sys v0.49.0 // indirect

go.sum

@@ -367,8 +367,8 @@ golang.org/x/crypto v0.58.0 h1:COkYLr4k7nDI4r6QtMML2AfzeIswWHAJthLS6K/M/54=
 golang.org/x/crypto v0.58.0/go.mod h1:Xh+kl5A+M0gMiWDytH2Hhr5Z7WbNCpv/eB/EFgfSk+w=
 golang.org/x/mod v0.42.0 h1:ICeL3t2L30ljSPHHxDnR4fqVcBt/zR2zJAb9juavmQg=
 golang.org/x/mod v0.42.0/go.mod h1:u5fzIrRInkL8MaO1cbRkUHT64v303YIT5KEqpNr7jRs=
-golang.org/x/net v0.61.0 h1:RR+6j/BTBCrggOLugxa+66R8FvoN6MHlzDlZIbN9YFw=
-golang.org/x/net v0.61.0/go.mod h1:WgDCOTH8iwtB66T4/GfI94AHo/hhdb7Hah9NDseQfiI=
+golang.org/x/net v0.62.0 h1:18ELalfM5+XNRwzM2nYqXsZQ9ByS1uhj2O/U7+5jqRY=
+golang.org/x/net v0.62.0/go.mod h1:WgDCOTH8iwtB66T4/GfI94AHo/hhdb7Hah9NDseQfiI=
 golang.org/x/oauth2 v0.37.0 h1:JUlcxA8oAtauLfiH8FX2/FkAWHAdi0QtGCGc+hofE98=
 golang.org/x/oauth2 v0.37.0/go.mod h1:IxwZNxUULJmpBFf9K/9NTMSIfZZuvuTy1gGxhigP/58=
 golang.org/x/sync v0.24.0 h1:iqiDrGoDgyuvAKrhLQthUEzU/R378N78gXbBjxed7oA=

internal/purchase/due_age_cap.go

@@ -0,0 +1,56 @@
+package purchase
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+)
+
+// DefaultMaxDueRowAge is how long after its due time the fire and process
+// sweeps may still buy a row when ManagerConfig.MaxDueRowAge is unset. It
+// matches config.LadderFireGrace, the same limit the ladder already applies to
+// its own scheduled tranches.
+const DefaultMaxDueRowAge = 24 * time.Hour
+
+// ErrDueRowTooOld marks a row the fire or process sweep refused because it was
+// due more than the due-row age cap ago (issue #712). The sweeps are opt-in on
+// most platforms, so the first tick after one is enabled can find a long
+// backlog of overdue rows whose recommendations and prices are stale; buying
+// them all at once is the outcome this cap prevents. The row ends failed with
+// no_submission set, so it can be retried deliberately from History.
+var ErrDueRowTooOld = errors.New("scheduled purchase skipped: it was due too long ago")
+
+// dueRowMaxAge is the sweeps' age cap, DefaultMaxDueRowAge when unset.
+func (m *Manager) dueRowMaxAge() time.Duration {
+	if m.maxDueRowAge > 0 {
+		return m.maxDueRowAge
+	}
+	return DefaultMaxDueRowAge
+}
+
+// dueAt is when the sweeps consider exec due: scheduled_execution_at for a
+// delayed (status "scheduled") purchase, which only the delayed-approve path
+// sets, and scheduled_date for every other row.
+func dueAt(exec *config.PurchaseExecution) time.Time {
+	if exec.ScheduledExecutionAt != nil {
+		return *exec.ScheduledExecutionAt
+	}
+	return exec.ScheduledDate
+}
+
+// dueRowAgeRefusal refuses exec when it was due more than maxAge before now.
+// Callers run it on the row their "running" claim returned, right before the
+// provider call: the claim already shut out every other executor, so a row
+// cannot age past the cap between this check and the purchase. The message is
+// what History shows as the row's error.
+func dueRowAgeRefusal(exec *config.PurchaseExecution, maxAge time.Duration, now time.Time) error {
+	due := dueAt(exec)
+	late := now.Sub(due)
+	if late <= maxAge {
+		return nil
+	}
+	return fmt.Errorf("%w: it was due at %s, %s ago, which is more than the %s limit (SCHEDULED_PURCHASE_MAX_OVERDUE); nothing was submitted, retry it if the purchase is still wanted",
+		ErrDueRowTooOld, due.UTC().Format(time.RFC3339), late.Round(time.Minute), maxAge)
+}

internal/purchase/due_age_cap_integration_test.go

@@ -0,0 +1,143 @@
+//go:build integration
+
+package purchase
+
+// Real PostgreSQL plus the ramp-step fixture's counting fake provider (labeled
+// fixture; nothing here reaches a cloud API). Issue #712: when the fire and
+// process sweeps are first scheduled, a backlog of overdue rows must not be
+// bought. Rows due more than the age cap ago end failed and unbought, with
+// the reason on the row, while rows inside the cap are still bought, each
+// exactly once, however many sweeps overlap.
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/email"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+// agedRow saves a row on accounts[0] that is due `late` ago: a "scheduled"
+// (delayed) purchase is due at scheduled_execution_at, every other row at
+// scheduled_date.
+func agedRow(ctx context.Context, t *testing.T, f *rampStepFixture, status string, late time.Duration) *config.PurchaseExecution {
+	t.Helper()
+	due := time.Now().Add(-late)
+	row := &config.PurchaseExecution{
+		PlanID: f.plan.ID, ExecutionID: uuid.New().String(), Status: status, StepNumber: 3,
+		ScheduledDate: due, CloudAccountID: &f.accounts[0].ID,
+		Recommendations: []config.RecommendationRecord{withAccount(rampStepRecommendation()[0], f.accounts[0].ID)},
+	}
+	if status == "scheduled" {
+		row.ScheduledDate = time.Now()
+		row.ScheduledExecutionAt = &due
+	}
+	require.NoError(t, f.store.SavePurchaseExecution(ctx, row))
+	return row
+}
+
+// countFailureEmails records each purchase-failed email by execution ID.
+func countFailureEmails(f *rampStepFixture) func() map[string]int {
+	var mu sync.Mutex
+	sent := map[string]int{}
+	f.email.On("SendPurchaseFailedNotification", mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			mu.Lock()
+			defer mu.Unlock()
+			sent[args.Get(1).(email.NotificationData).ExecutionID]++
+		}).Return(nil).Maybe()
+	return func() map[string]int {
+		mu.Lock()
+		defer mu.Unlock()
+		out := make(map[string]int, len(sent))
+		for k, v := range sent {
+			out[k] = v
+		}
+		return out
+	}
+}
+
+// Overlapping fire and process sweeps over a backlog that holds one overdue
+// and one fresh row of each kind. Before the cap, every overdue row was bought.
+func TestDueRowAgeCap_OverlappingSweepsBuyOnlyRowsInsideTheCap(t *testing.T) {
+	ctx := context.Background()
+	f := newRampStepFixture(ctx, t)
+	f.repairAccounts(ctx, t)
+	emails := countFailureEmails(f)
+
+	overdue := []*config.PurchaseExecution{
+		agedRow(ctx, t, f, "notified", 48*time.Hour),
+		agedRow(ctx, t, f, "scheduled", 48*time.Hour),
+	}
+	fresh := []*config.PurchaseExecution{
+		agedRow(ctx, t, f, "notified", time.Hour),
+		agedRow(ctx, t, f, "scheduled", time.Hour),
+	}
+
+	const sweeps = 4
+	var mu sync.Mutex
+	fired := 0
+	var wg sync.WaitGroup
+	for i := 0; i < sweeps; i++ {
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			_, _ = f.manager.ProcessScheduledPurchases(ctx)
+		}()
+		go func() {
+			defer wg.Done()
+			if res, err := f.manager.FireScheduledDelayedPurchases(ctx); err == nil {
+				mu.Lock()
+				fired += res.Fired
+				mu.Unlock()
+			}
+		}()
+	}
+	wg.Wait()
+
+	assert.Equal(t, len(fresh), f.purchaseCount(), "only the rows inside the cap reach the provider, each once")
+	assert.Equal(t, 1, fired, "the fire sweeps fire the fresh delayed purchase once and the overdue one never")
+	sent := emails()
+	for _, row := range overdue {
+		got := rowStatus(ctx, t, f, row.ExecutionID)
+		assert.Equal(t, config.StatusFailed, got.Status, "overdue %s row", row.Status)
+		assert.True(t, got.NoSubmission, "overdue %s row is proven unbought", row.Status)
+		assert.Contains(t, got.Error, "due too long ago", "overdue %s row says why it was skipped", row.Status)
+		assert.Equal(t, 1, sent[row.ExecutionID], "operators hear about overdue %s row exactly once", row.Status)
+	}
+	for _, row := range fresh {
+		assert.Equal(t, "completed", rowStatus(ctx, t, f, row.ExecutionID).Status, "fresh %s row", row.Status)
+		assert.Zero(t, sent[row.ExecutionID], "fresh %s row sends no failure email", row.Status)
+	}
+}
+
+// A later tick finds nothing left to do: the refused rows are terminal and
+// are neither bought nor reported again.
+func TestDueRowAgeCap_RefusedRowsStayRefused(t *testing.T) {
+	ctx := context.Background()
+	f := newRampStepFixture(ctx, t)
+	f.repairAccounts(ctx, t)
+	emails := countFailureEmails(f)
+	notified := agedRow(ctx, t, f, "notified", 48*time.Hour)
+	scheduled := agedRow(ctx, t, f, "scheduled", 48*time.Hour)
+
+	for tick := 0; tick < 2; tick++ {
+		_, err := f.manager.ProcessScheduledPurchases(ctx)
+		require.NoError(t, err)
+		_, err = f.manager.FireScheduledDelayedPurchases(ctx)
+		require.NoError(t, err)
+	}
+
+	assert.Zero(t, f.purchaseCount())
+	sent := emails()
+	for _, row := range []*config.PurchaseExecution{notified, scheduled} {
+		assert.Equal(t, config.StatusFailed, rowStatus(ctx, t, f, row.ExecutionID).Status)
+		assert.Equal(t, 1, sent[row.ExecutionID])
+	}
+}