diff --git a/.env.example b/.env.example index 4760b245..e0106e4c 100644 --- a/.env.example +++ b/.env.example @@ -120,6 +120,7 @@ DASHBOARD_URL=http://localhost:3000 # Optional: tunables # --------------------------------------------------------------------- # CUDLY_RECOMMENDATION_CACHE_TTL=15m +# CUDLY_STALE_RECOMMENDATION_RETENTION_DAYS=30 # prune recs of (provider, account) pairs not collected for this many days # RECOMMENDATION_SCHEDULE=cron(0 2 * * ? *) # Terraform sets it; drives next_collection_at # CUDLY_MAX_ACCOUNT_PARALLELISM=8 # DEFAULT_PAYMENT_OPTION=no-upfront diff --git a/internal/config/constants.go b/internal/config/constants.go index ea355f43..afd12c6e 100644 --- a/internal/config/constants.go +++ b/internal/config/constants.go @@ -14,6 +14,16 @@ const ( // DefaultExecutionTTLDays is how long execution records are kept. DefaultExecutionTTLDays = 30 + // DefaultStaleRecommendationRetentionDays is how long the cached + // recommendations of a (provider, account) pair survive once that pair + // stops collecting (a removed provider, a disabled account, credentials + // that keep failing). Collection runs daily on every cloud (#1580), so + // this is about 30 missed collections in a row: long enough to ride out + // an outage or an expired credential nobody fixed for a few weeks, short + // enough that a pair that is gone for good stops showing within a month. + // PruneStaleRecommendations applies it after every collection (#120). + DefaultStaleRecommendationRetentionDays = 30 + // DefaultMaxRecommendationsInEmail is the max recommendations shown in email notifications. DefaultMaxRecommendationsInEmail = 10 diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index f34129c3..01947b1c 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -677,6 +677,11 @@ type StoreInterface interface { // SuccessfulCollect for the per-row semantics. There is deliberately no // unscoped "replace everything" method (issue #159). UpsertRecommendations(ctx context.Context, collectedAt time.Time, recs []RecommendationRecord, successfulCollects []SuccessfulCollect) error + // PruneStaleRecommendations deletes the rows of every (provider, + // account) pair with no row collected in the last retentionDays days, + // leaving every pair that collected inside the window untouched, and + // returns the number of rows deleted (issue #120). + PruneStaleRecommendations(ctx context.Context, retentionDays int) (int64, error) ListStoredRecommendations(ctx context.Context, filter RecommendationFilter) ([]RecommendationRecord, error) // ListStoredRecommendationsChunk is ListStoredRecommendations for a // keyset walk: at most limit rows after the position after, in diff --git a/internal/config/recommendations_scoped_delete_guard_test.go b/internal/config/recommendations_scoped_delete_guard_test.go index ecdf5638..b92473d8 100644 --- a/internal/config/recommendations_scoped_delete_guard_test.go +++ b/internal/config/recommendations_scoped_delete_guard_test.go @@ -25,9 +25,11 @@ var ( // TestNoUnscopedRecommendationsDelete guards #159. ReplaceRecommendations ran // `DELETE FROM recommendations` with no WHERE, so any caller handing it a // partial or empty snapshot would have wiped every account's rows, not only -// its own. It had no production caller and was removed. The only production -// delete left on the table is UpsertRecommendations' eviction, which is -// scoped to the (provider, account) pairs that collected successfully. +// its own. It had no production caller and was removed. The production +// deletes left on the table are UpsertRecommendations' eviction, which is +// scoped to the (provider, account) pairs that collected successfully, and +// PruneStaleRecommendations, which is scoped to the pairs with no row +// collected inside the retention window (issue #120). // // This scans every non-test Go file under internal/ and cmd/ for a string // literal that deletes from recommendations without a WHERE in the same diff --git a/internal/config/store_postgres_recommendations_prune.go b/internal/config/store_postgres_recommendations_prune.go new file mode 100644 index 00000000..4785fec3 --- /dev/null +++ b/internal/config/store_postgres_recommendations_prune.go @@ -0,0 +1,74 @@ +package config + +import ( + "context" + "fmt" +) + +// PruneStaleRecommendations deletes the cached recommendations of every +// (provider, account) pair that has not collected for retentionDays, and +// returns how many rows it deleted (issue #120). +// +// UpsertRecommendations only evicts rows of the pairs that collected +// successfully in the current run, so the rows of a pair that never collects +// again (its provider was disabled, its account disabled, or its credentials +// keep failing) would otherwise stay on the Recommendations page forever. +// Registered accounts that are deleted outright are already covered: the +// cloud_account_id FK cascades. +// +// A pair's newest row is its last collection: every collection stamps the +// rows it upserts with its own collected_at, and a successful one evicts the +// pair's older rows. A pair counts as stale only when it has no row at all +// collected inside the window, and only rows of such pairs are deleted. A pair +// with even one recent row keeps every row, however old the rest are (a +// partial collection leaves the failed services' rows behind on purpose). +// +// Concurrent collections: the table lock below (SHARE ROW EXCLUSIVE conflicts +// with the ROW EXCLUSIVE lock every INSERT, UPDATE and DELETE takes, and with +// itself) makes the prune wait for any UpsertRecommendations transaction still +// in flight, and makes collections that start meanwhile wait for the prune. +// The DELETE's snapshot is taken after the lock is granted, so a collection +// that committed fresh rows for a pair always counts, and no collection can +// commit between the pair check and the delete. Readers take ACCESS SHARE, +// which this lock does not block. The prune holds no other lock before +// requesting this one, and nothing in the codebase row-locks recommendations +// without writing them, so the wait cannot deadlock. +// +// The row-level collected_at bound is implied by the NOT EXISTS (every row of +// a stale pair is older than the cutoff); it is kept so the planner can drive +// the delete from recommendations_collected_idx rather than scan the table. +func (s *PostgresStore) PruneStaleRecommendations(ctx context.Context, retentionDays int) (int64, error) { + if retentionDays <= 0 { + return 0, fmt.Errorf("prune stale recommendations: retention must be at least 1 day, got %d", retentionDays) + } + + tx, err := s.db.Begin(ctx) + if err != nil { + return 0, fmt.Errorf("prune stale recommendations: begin tx: %w", err) + } + defer rollbackTx(ctx, tx, "PruneStaleRecommendations") + + if _, err = tx.Exec(ctx, `LOCK TABLE recommendations IN SHARE ROW EXCLUSIVE MODE`); err != nil { + return 0, fmt.Errorf("prune stale recommendations: lock table: %w", err) + } + + tag, err := tx.Exec(ctx, ` + DELETE FROM recommendations r + WHERE r.collected_at < NOW() - INTERVAL '1 day' * $1 + AND NOT EXISTS ( + SELECT 1 + FROM recommendations fresh + WHERE fresh.provider = r.provider + AND fresh.account_key = r.account_key + AND fresh.collected_at >= NOW() - INTERVAL '1 day' * $1 + ) + `, retentionDays) + if err != nil { + return 0, fmt.Errorf("prune stale recommendations: delete: %w", err) + } + + if err = tx.Commit(ctx); err != nil { + return 0, fmt.Errorf("prune stale recommendations: commit: %w", err) + } + return tag.RowsAffected(), nil +} diff --git a/internal/config/store_postgres_recommendations_prune_integration_test.go b/internal/config/store_postgres_recommendations_prune_integration_test.go new file mode 100644 index 00000000..4721bda2 --- /dev/null +++ b/internal/config/store_postgres_recommendations_prune_integration_test.go @@ -0,0 +1,267 @@ +//go:build integration +// +build integration + +package config_test + +import ( + "context" + "sort" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// pruneRetentionDays is the retention every prune test runs with; the seeded +// ages below are chosen around it. +const pruneRetentionDays = 30 + +// azureAccountRec is an Azure rec owned by a registered account. +func azureAccountRec(id, accountID, resourceType string) config.RecommendationRecord { + acct := accountID + return config.RecommendationRecord{ + ID: id, + CloudAccountID: &acct, + Provider: "azure", + Service: "vm", + Region: "eastus", + ResourceType: resourceType, + Savings: 10, + Count: 1, + Term: 12, + Payment: "no-upfront", + } +} + +// ambientRec is a rec with no registered account (account_key = zero UUID). +func ambientRec(id, provider, resourceType string) config.RecommendationRecord { + return config.RecommendationRecord{ + ID: id, + Provider: provider, + Service: "vm", + Region: "eastus", + ResourceType: resourceType, + Savings: 10, + Count: 1, + Term: 12, + Payment: "no-upfront", + } +} + +// seedAt writes recs stamped with collectedAt and evicts nothing (no +// successful collects), so each call adds rows of the given age. +func seedAt(ctx context.Context, t *testing.T, store *config.PostgresStore, collectedAt time.Time, recs ...config.RecommendationRecord) { + t.Helper() + require.NoError(t, store.UpsertRecommendations(ctx, collectedAt, recs, nil)) +} + +// storedRecIDs returns the sorted payload ids of every stored row. +func storedRecIDs(ctx context.Context, t *testing.T, store *config.PostgresStore) []string { + t.Helper() + recs, err := store.ListStoredRecommendations(ctx, config.RecommendationFilter{}) + require.NoError(t, err) + ids := make([]string, 0, len(recs)) + for i := range recs { + ids = append(ids, recs[i].ID) + } + sort.Strings(ids) + return ids +} + +// TestPostgresStore_PruneStaleRecommendations_PairScope pins issue #120: only +// the rows of a (provider, account) pair with no row inside the window go, +// and a pair with one recent row keeps all of its rows, however old. +func TestPostgresStore_PruneStaleRecommendations_PairScope(t *testing.T) { + ctx := context.Background() + store := setupRecommendationsStore(ctx, t) + + staleAcct := uuid.NewString() + freshAcct := uuid.NewString() + edgeAcct := uuid.NewString() + seedRecommendationCloudAccount(ctx, t, store, staleAcct, "azure", "sub-120-stale") + seedRecommendationCloudAccount(ctx, t, store, freshAcct, "azure", "sub-120-fresh") + seedRecommendationCloudAccount(ctx, t, store, edgeAcct, "azure", "sub-120-edge") + + now := time.Now().UTC() + old := now.AddDate(0, 0, -(pruneRetentionDays + 10)) + justOld := now.AddDate(0, 0, -(pruneRetentionDays + 1)) + recent := now.Add(-time.Hour) + insideWindow := now.AddDate(0, 0, -(pruneRetentionDays - 1)) + + // Stale registered pair: every row is outside the window. + seedAt(ctx, t, store, old, azureAccountRec("stale-1", staleAcct, "D2"), azureAccountRec("stale-2", staleAcct, "D4")) + // Fresh registered pair with mixed ages: one recent row keeps the others. + seedAt(ctx, t, store, recent, azureAccountRec("fresh-new", freshAcct, "D2")) + seedAt(ctx, t, store, old, azureAccountRec("fresh-old", freshAcct, "D4")) + seedAt(ctx, t, store, justOld, azureAccountRec("fresh-older", freshAcct, "D8")) + // A pair whose newest row is a day inside the window is not stale yet. + seedAt(ctx, t, store, insideWindow, azureAccountRec("edge-1", edgeAcct, "D2")) + // Ambient pairs: the fresh azure ambient pair must not shield the stale + // aws ambient pair (provider is part of the pair) nor the stale azure + // registered pair above (the account is part of the pair). + seedAt(ctx, t, store, old, ambientRec("aws-ambient-old", "aws", "m5.large")) + seedAt(ctx, t, store, recent, ambientRec("azure-ambient-new", "azure", "D16")) + + deleted, err := store.PruneStaleRecommendations(ctx, pruneRetentionDays) + require.NoError(t, err) + assert.Equal(t, int64(3), deleted, "two stale registered rows and one stale ambient row") + assert.Equal(t, + []string{"azure-ambient-new", "edge-1", "fresh-new", "fresh-old", "fresh-older"}, + storedRecIDs(ctx, t, store)) + + // Idempotent: nothing left to prune. + deleted, err = store.PruneStaleRecommendations(ctx, pruneRetentionDays) + require.NoError(t, err) + assert.Zero(t, deleted) +} + +// A non-positive retention is refused rather than read as "everything is +// stale". +func TestPostgresStore_PruneStaleRecommendations_RejectsNonPositiveRetention(t *testing.T) { + ctx := context.Background() + store := setupRecommendationsStore(ctx, t) + seedAt(ctx, t, store, time.Now().UTC().AddDate(0, 0, -400), ambientRec("ancient", "aws", "m5.large")) + + for _, days := range []int{0, -1} { + deleted, err := store.PruneStaleRecommendations(ctx, days) + require.Error(t, err, "retention %d", days) + assert.Zero(t, deleted) + } + assert.Equal(t, []string{"ancient"}, storedRecIDs(ctx, t, store)) +} + +// waitForLockWaiters blocks until at least n backends of this test's database +// are waiting on a heavyweight lock. +func waitForLockWaiters(ctx context.Context, t *testing.T, db *database.Connection, n int, what string) { + t.Helper() + require.Eventually(t, func() bool { + var waiting int + if err := db.QueryRow(ctx, ` + SELECT count(*) FROM pg_stat_activity + WHERE datname = current_database() + AND wait_event_type = 'Lock' + `).Scan(&waiting); err != nil { + return false + } + return waiting >= n + }, 10*time.Second, 20*time.Millisecond, what) +} + +// TestPostgresStore_PruneStaleRecommendations_ConcurrentCollection runs the +// prune while a collection that refreshes a stale pair is still in flight. The +// collection has already upserted one of the pair's rows (holding its row +// lock) and is held open on the recommendations_state row lock. The prune must +// wait for it and then treat the pair as fresh: it keeps both the refreshed +// row and the pair's other, older row, while a pair nobody collected is still +// pruned. +func TestPostgresStore_PruneStaleRecommendations_ConcurrentCollection(t *testing.T) { + ctx := context.Background() + db := testhelpers.RequireMigratedDatabase(ctx, t, getMigrationsPath()).DB + store := config.NewPostgresStore(db) + + pairAcct := uuid.NewString() + goneAcct := uuid.NewString() + seedRecommendationCloudAccount(ctx, t, store, pairAcct, "azure", "sub-120-pair") + seedRecommendationCloudAccount(ctx, t, store, goneAcct, "azure", "sub-120-gone") + + old := time.Now().UTC().AddDate(0, 0, -(pruneRetentionDays + 10)) + seedAt(ctx, t, store, old, + azureAccountRec("pair-refreshed", pairAcct, "D2"), + azureAccountRec("pair-untouched", pairAcct, "D4"), + azureAccountRec("gone-1", goneAcct, "D2")) + + // Hold the state row so the collection below blocks on its final write, + // after its upsert has taken the row lock on pair-refreshed. + hold, err := db.Begin(ctx) + require.NoError(t, err) + t.Cleanup(func() { _ = hold.Rollback(context.Background()) }) + _, err = hold.Exec(ctx, `SELECT 1 FROM recommendations_state WHERE id = 1 FOR UPDATE`) + require.NoError(t, err) + + // A partial collection for the pair: it refreshes one row and, because the + // pair is not in successfulCollects, leaves the other row in place. + collectedAt := time.Now().UTC() + collectDone := make(chan error, 1) + go func() { + collectDone <- store.UpsertRecommendations(ctx, collectedAt, + []config.RecommendationRecord{azureAccountRec("pair-refreshed", pairAcct, "D2")}, nil) + }() + waitForLockWaiters(ctx, t, db, 1, "collection never blocked on the held state row") + + type pruneResult struct { + deleted int64 + err error + } + pruneDone := make(chan pruneResult, 1) + go func() { + n, perr := store.PruneStaleRecommendations(ctx, pruneRetentionDays) + pruneDone <- pruneResult{n, perr} + }() + waitForLockWaiters(ctx, t, db, 2, "prune never waited on the in-flight collection") + + require.NoError(t, hold.Commit(ctx)) + + select { + case cerr := <-collectDone: + require.NoError(t, cerr) + case <-time.After(10 * time.Second): + t.Fatal("collection did not finish after the state row was released") + } + var res pruneResult + select { + case res = <-pruneDone: + require.NoError(t, res.err) + case <-time.After(10 * time.Second): + t.Fatal("prune did not finish after the collection committed") + } + + assert.Equal(t, int64(1), res.deleted, "only the uncollected pair's row is pruned") + assert.Equal(t, []string{"pair-refreshed", "pair-untouched"}, storedRecIDs(ctx, t, store)) + + var stamped time.Time + require.NoError(t, db.QueryRow(ctx, + `SELECT collected_at FROM recommendations WHERE payload->>'id' = 'pair-refreshed'`).Scan(&stamped)) + assert.WithinDuration(t, collectedAt, stamped, time.Second, "the refreshed row keeps the collection's timestamp") +} + +// TestPostgresStore_PruneStaleRecommendations_RacingCollections races prunes +// against collections that refresh a stale pair, in no fixed order. Whichever +// commits first, the refreshed row must survive: a prune that runs first +// deletes the old rows and the collection re-inserts, and a prune that runs +// second sees the pair as fresh. +func TestPostgresStore_PruneStaleRecommendations_RacingCollections(t *testing.T) { + ctx := context.Background() + store := setupRecommendationsStore(ctx, t) + acct := uuid.NewString() + seedRecommendationCloudAccount(ctx, t, store, acct, "azure", "sub-120-race") + + for round := 0; round < 20; round++ { + old := time.Now().UTC().AddDate(0, 0, -(pruneRetentionDays + 10)) + seedAt(ctx, t, store, old, azureAccountRec("race-row", acct, "D2")) + + start := make(chan struct{}) + collectDone := make(chan error, 1) + pruneDone := make(chan error, 1) + go func() { + <-start + collectDone <- store.UpsertRecommendations(ctx, time.Now().UTC(), + []config.RecommendationRecord{azureAccountRec("race-row", acct, "D2")}, + []config.SuccessfulCollect{{Provider: "azure", CloudAccountID: &acct}}) + }() + go func() { + <-start + _, perr := store.PruneStaleRecommendations(ctx, pruneRetentionDays) + pruneDone <- perr + }() + close(start) + require.NoError(t, <-collectDone, "round %d", round) + require.NoError(t, <-pruneDone, "round %d", round) + + require.Equal(t, []string{"race-row"}, storedRecIDs(ctx, t, store), "round %d: refreshed row lost", round) + } +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index ce62f622..b6d415b0 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1830,6 +1830,19 @@ func (m *MockConfigStore) UpsertRecommendations(ctx context.Context, collectedAt return args.Error(0) } +func (m *MockConfigStore) PruneStaleRecommendations(ctx context.Context, retentionDays int) (int64, error) { + m.record("PruneStaleRecommendations", ctx, retentionDays) + if !isExpected(&m.Mock, "PruneStaleRecommendations") { + return 0, nil + } + args := m.Called(ctx, retentionDays) + v, ok := args.Get(0).(int64) + if !ok { + m.reporter.wrongReturnType("PruneStaleRecommendations", "int64", args.Get(0)) + } + return v, args.Error(1) +} + func (m *MockConfigStore) ListStoredRecommendations(ctx context.Context, filter config.RecommendationFilter) ([]config.RecommendationRecord, error) { m.record("ListStoredRecommendations", ctx, filter) if !isExpected(&m.Mock, "ListStoredRecommendations") { diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index ecc9fb21..de871873 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -438,6 +438,12 @@ func (s *Scheduler) clearCollectionStartedBestEffort(ctx context.Context, ownerT // re-records it straight after (issue #87). The two writes mirror the // existing error-column pattern and are not atomic: if the second write // fails, the run is logged and reads as complete until the next collection. +// +// After a successful write it prunes the rows of (provider, account) pairs +// that have not collected for CUDLY_STALE_RECOMMENDATION_RETENTION_DAYS +// (default 30), best-effort, so a removed provider or a disabled account +// stops showing instead of staying forever (issue #120). A failed write skips +// the prune. func (s *Scheduler) persistCollection(ctx context.Context, recs []config.RecommendationRecord, successfulCollects []config.SuccessfulCollect, failedProviders map[string]string, partial *config.RecommendationsCollectionPartial) { if err := s.config.UpsertRecommendations(ctx, time.Now().UTC(), recs, successfulCollects); err != nil { logging.Errorf("Failed to persist recommendations: %v", err) @@ -458,6 +464,8 @@ func (s *Scheduler) persistCollection(ctx context.Context, recs []config.Recomme logging.Errorf("Failed to record partial collection: %v", err) } } + + s.pruneStaleRecommendationsBestEffort(ctx) } // joinProviderErrors formats a "provider: err; provider: err" string for diff --git a/internal/scheduler/stale_recommendations.go b/internal/scheduler/stale_recommendations.go new file mode 100644 index 00000000..5b204bab --- /dev/null +++ b/internal/scheduler/stale_recommendations.go @@ -0,0 +1,55 @@ +package scheduler + +import ( + "context" + "os" + "strconv" + "strings" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// staleRecommendationRetentionEnvVar overrides +// config.DefaultStaleRecommendationRetentionDays: the number of days a +// (provider, account) pair may go without collecting before its cached +// recommendations are deleted (issue #120). +const staleRecommendationRetentionEnvVar = "CUDLY_STALE_RECOMMENDATION_RETENTION_DAYS" + +// staleRecommendationRetentionDaysFromEnv reads +// CUDLY_STALE_RECOMMENDATION_RETENTION_DAYS on every call, so a changed value +// takes effect on the next collection without a restart. Unset uses the +// default; a value that is not a positive whole number of days logs a warning +// and uses the default too, so a typo can never turn into "delete everything" +// (0) or an error that skips the prune for good. +func staleRecommendationRetentionDaysFromEnv() int { + raw := strings.TrimSpace(os.Getenv(staleRecommendationRetentionEnvVar)) + if raw == "" { + return config.DefaultStaleRecommendationRetentionDays + } + days, err := strconv.Atoi(raw) + if err != nil || days <= 0 { + logging.Warnf("%s invalid (%q, want a positive number of days); using default %d", + staleRecommendationRetentionEnvVar, raw, config.DefaultStaleRecommendationRetentionDays) + return config.DefaultStaleRecommendationRetentionDays + } + return days +} + +// pruneStaleRecommendationsBestEffort deletes the cached rows of the +// (provider, account) pairs that have not collected within the retention +// window. It runs after every persisted collection, which is daily on every +// cloud by default, rather than in the cleanup sweep, which is opt-in on every +// platform. A failure is logged and never fails the collection: the rows it +// would have deleted are simply kept until the next run. +func (s *Scheduler) pruneStaleRecommendationsBestEffort(ctx context.Context) { + days := staleRecommendationRetentionDaysFromEnv() + deleted, err := s.config.PruneStaleRecommendations(ctx, days) + if err != nil { + logging.Errorf("Failed to prune recommendations of pairs not collected for %d days: %v", days, err) + return + } + if deleted > 0 { + logging.Infof("Pruned %d recommendations of (provider, account) pairs not collected for %d days", deleted, days) + } +} diff --git a/internal/scheduler/stale_recommendations_test.go b/internal/scheduler/stale_recommendations_test.go new file mode 100644 index 00000000..11dd09a2 --- /dev/null +++ b/internal/scheduler/stale_recommendations_test.go @@ -0,0 +1,106 @@ +package scheduler + +import ( + "context" + "errors" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" +) + +// TestStaleRecommendationRetentionDaysFromEnv pins the parse rules: unset and +// anything that is not a positive whole number of days fall back to the +// default, so a typo can never become a zero-day retention. +func TestStaleRecommendationRetentionDaysFromEnv(t *testing.T) { + cases := []struct { + name string + raw string + want int + }{ + {"unset", "", config.DefaultStaleRecommendationRetentionDays}, + {"valid", "45", 45}, + {"surrounding space", " 7 ", 7}, + {"one day", "1", 1}, + {"zero", "0", config.DefaultStaleRecommendationRetentionDays}, + {"negative", "-3", config.DefaultStaleRecommendationRetentionDays}, + {"duration syntax", "720h", config.DefaultStaleRecommendationRetentionDays}, + {"garbage", "thirty", config.DefaultStaleRecommendationRetentionDays}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Setenv(staleRecommendationRetentionEnvVar, tc.raw) + assert.Equal(t, tc.want, staleRecommendationRetentionDaysFromEnv()) + }) + } +} + +// The default is 30 days: about 30 missed daily collections (#1580). +func TestDefaultStaleRecommendationRetentionDays(t *testing.T) { + assert.Equal(t, 30, config.DefaultStaleRecommendationRetentionDays) +} + +// A persisted collection prunes stale pairs with the configured retention. +func TestScheduler_persistCollection_PrunesStalePairs(t *testing.T) { + t.Setenv(staleRecommendationRetentionEnvVar, "") + ctx := context.Background() + mockStore := new(MockConfigStore) + recs := []config.RecommendationRecord{{ID: "r1", Provider: "aws"}} + successful := []config.SuccessfulCollect{{Provider: "aws"}} + mockStore.On("UpsertRecommendations", ctx, mock.Anything, recs, successful).Return(nil) + mockStore.On("PruneStaleRecommendations", ctx, config.DefaultStaleRecommendationRetentionDays).Return(int64(4), nil) + + s := &Scheduler{config: mockStore} + s.persistCollection(ctx, recs, successful, nil, nil) + + mockStore.AssertExpectations(t) + mockStore.AssertNumberOfCalls(t, "PruneStaleRecommendations", 1) +} + +// The env override reaches the store. +func TestScheduler_persistCollection_PruneUsesEnvRetention(t *testing.T) { + t.Setenv(staleRecommendationRetentionEnvVar, "60") + ctx := context.Background() + mockStore := new(MockConfigStore) + mockStore.On("UpsertRecommendations", ctx, mock.Anything, mock.Anything, mock.Anything).Return(nil) + mockStore.On("PruneStaleRecommendations", ctx, 60).Return(int64(0), nil) + + s := &Scheduler{config: mockStore} + s.persistCollection(ctx, nil, nil, nil, nil) + + mockStore.AssertExpectations(t) +} + +// A prune failure is logged and does not undo or fail the collection: the +// partial marker is still recorded before it and nothing else is called after. +func TestScheduler_persistCollection_PruneErrorIsBestEffort(t *testing.T) { + t.Setenv(staleRecommendationRetentionEnvVar, "") + ctx := context.Background() + mockStore := new(MockConfigStore) + partial := &config.RecommendationsCollectionPartial{} + mockStore.On("UpsertRecommendations", ctx, mock.Anything, mock.Anything, mock.Anything).Return(nil) + mockStore.On("SetRecommendationsCollectionPartial", ctx, partial).Return(nil) + mockStore.On("PruneStaleRecommendations", ctx, config.DefaultStaleRecommendationRetentionDays). + Return(int64(0), errors.New("lock timeout")) + + s := &Scheduler{config: mockStore} + s.persistCollection(ctx, nil, nil, nil, partial) + + mockStore.AssertExpectations(t) + mockStore.AssertNotCalled(t, "SetRecommendationsCollectionError", mock.Anything, mock.Anything) +} + +// A failed write skips the prune: the database is the thing failing, and a +// run that wrote nothing has no fresher data to justify deleting older rows. +func TestScheduler_persistCollection_UpsertFailureSkipsPrune(t *testing.T) { + ctx := context.Background() + mockStore := new(MockConfigStore) + mockStore.On("UpsertRecommendations", ctx, mock.Anything, mock.Anything, mock.Anything).Return(errors.New("db down")) + mockStore.On("SetRecommendationsCollectionError", ctx, mock.Anything).Return(nil) + + s := &Scheduler{config: mockStore} + s.persistCollection(ctx, nil, nil, nil, nil) + + mockStore.AssertNotCalled(t, "PruneStaleRecommendations") +} diff --git a/specs/recommendations-cache.md b/specs/recommendations-cache.md index df9013f6..dce3e628 100644 --- a/specs/recommendations-cache.md +++ b/specs/recommendations-cache.md @@ -29,6 +29,16 @@ is scoped to providers that actually succeeded in the current run, so rows from a provider whose collect failed stay visible instead of being blanked out. +Those kept rows are not kept forever. After every persisted collection, +`PruneStaleRecommendations` deletes the rows of each (provider, account) +pair that has no row collected in the last +`CUDLY_STALE_RECOMMENDATION_RETENTION_DAYS` days: a disabled provider, a +disabled account or credentials that keep failing (issue #120). A pair +with even one row inside the window keeps all of its rows. The prune takes +a `SHARE ROW EXCLUSIVE` lock on `recommendations`, so it waits for any +in-flight collection to commit and never judges a pair on a snapshot older +than that collection. + Per-account cloud-API calls within each provider are fanned out via `errgroup.SetLimit`, bounded by `CUDLY_MAX_ACCOUNT_PARALLELISM` (default 20). Partial failures are logged and the run continues. @@ -70,6 +80,11 @@ Lambda deploys rely on scheduled + manual refresh only. - `CUDLY_RECOMMENDATION_CACHE_TTL` (default `6h`) — parsed by `time.ParseDuration`. Invalid values fall back to the default with a warning. Controls when in-request stale-while-revalidate fires. +- `CUDLY_STALE_RECOMMENDATION_RETENTION_DAYS` (default `30`): how many + days a (provider, account) pair may go without collecting before its + cached rows are pruned. Collection is daily by default, so this is about + 30 missed collections. Read on every collection; a value that is not a + positive whole number falls back to the default with a warning. - `CUDLY_MAX_ACCOUNT_PARALLELISM` (default `20`) — caps per-account concurrency inside each provider phase. Shared with the purchase manager so both honour the same operator override.