diff --git a/internal/jobs/deploy_lifecycle_coverage_test.go b/internal/jobs/deploy_lifecycle_coverage_test.go index 787c949..7c68b44 100644 --- a/internal/jobs/deploy_lifecycle_coverage_test.go +++ b/internal/jobs/deploy_lifecycle_coverage_test.go @@ -38,6 +38,7 @@ import ( "database/sql" "encoding/json" "errors" + "fmt" "net" "net/http" "net/http/httptest" @@ -1793,7 +1794,7 @@ func TestOrphanSweep_Pass5_StackIDsQueryError(t *testing.T) { defer db.Close() mock.ExpectQuery(`SELECT d.app_id, d.status, t.status, d.created_at\s+FROM deployments d\s+JOIN teams t`). WillReturnRows(sqlmock.NewRows([]string{"app_id", "d_status", "t_status", "created_at"})) - mock.ExpectQuery(`SELECT id::text FROM stacks`). + mock.ExpectQuery(`SELECT id::text\s+FROM stacks`). WillReturnError(errors.New("conn lost")) lister := newFakeNamespaceLister().withStackNamespaces(ExpireStacksNamespacePrefix + "stack-1") @@ -1890,7 +1891,7 @@ func TestOrphanSweep_Pass5_DeleteFails(t *testing.T) { orphanNS := ExpireStacksNamespacePrefix + "willfail" mock.ExpectQuery(`SELECT d.app_id, d.status, t.status, d.created_at\s+FROM deployments d\s+JOIN teams t`). WillReturnRows(sqlmock.NewRows([]string{"app_id", "d_status", "t_status", "created_at"})) - mock.ExpectQuery(`SELECT id::text FROM stacks`). + mock.ExpectQuery(`SELECT id::text\s+FROM stacks`). WillReturnRows(sqlmock.NewRows([]string{"id"})) lister := newFakeNamespaceLister().withStackNamespaces(orphanNS) @@ -2121,7 +2122,7 @@ func TestOrphanSweep_FetchLiveStackIDs_ScanError(t *testing.T) { t.Fatalf("sqlmock.New: %v", err) } defer db.Close() - mock.ExpectQuery(`SELECT id::text FROM stacks`). + mock.ExpectQuery(`SELECT id::text\s+FROM stacks`). WillReturnRows(sqlmock.NewRows([]string{"id", "extra"}). AddRow("id", "extra")) @@ -2132,6 +2133,113 @@ func TestOrphanSweep_FetchLiveStackIDs_ScanError(t *testing.T) { } } +// TestOrphanSweep_FetchLiveStackIDs_KeysetPagination proves the bug-bash +// 2026-06-03 fix: fetchLiveStackIDs no longer issues one unbounded SELECT but +// streams the live stack ids in keyset-paginated batches. The first page is +// FULL (== orphanLiveIDsBatchLimit rows) so the loop must issue a SECOND query +// whose cursor ($1) is the last id of page 1; the second page is short, ending +// the loop. The assertion: every id from BOTH pages lands in the returned set, +// AND the second query's keyset arg equals page 1's tail (proving the cursor +// advanced rather than re-scanning from the start). +func TestOrphanSweep_FetchLiveStackIDs_KeysetPagination(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + + // Page 1: exactly orphanLiveIDsBatchLimit rows, zero-padded so they sort + // lexicographically in the same order we add them. The last id is the + // keyset cursor the second query must carry. + page1 := sqlmock.NewRows([]string{"id"}) + var lastPage1ID string + for i := 0; i < orphanLiveIDsBatchLimit; i++ { + id := fmt.Sprintf("stack-%06d", i) + page1.AddRow(id) + lastPage1ID = id + } + queryRE := `SELECT id::text\s+FROM stacks\s+WHERE id::text > \$1\s+ORDER BY id::text ASC\s+LIMIT \$2` + mock.ExpectQuery(queryRE). + WithArgs("", orphanLiveIDsBatchLimit). + WillReturnRows(page1) + // Page 2: short (2 rows < limit) → loop terminates. The cursor MUST be + // page 1's tail id. + mock.ExpectQuery(queryRE). + WithArgs(lastPage1ID, orphanLiveIDsBatchLimit). + WillReturnRows(sqlmock.NewRows([]string{"id"}). + AddRow("stack-overflow-a"). + AddRow("stack-overflow-b")) + + w := &OrphanSweepReconciler{db: db} + got, err := w.fetchLiveStackIDs(context.Background()) + if err != nil { + t.Fatalf("fetchLiveStackIDs: %v", err) + } + wantCount := orphanLiveIDsBatchLimit + 2 + if len(got) != wantCount { + t.Fatalf("live id count = %d; want %d (both pages merged)", len(got), wantCount) + } + if !got[lastPage1ID] { + t.Errorf("page 1 tail id %q missing from set", lastPage1ID) + } + if !got["stack-overflow-a"] || !got["stack-overflow-b"] { + t.Errorf("page 2 ids missing from set: %v", got) + } + if err := mock.ExpectationsWereMet(); err != nil { + // Unmet expectation here = the second keyset query never fired (the + // loop didn't paginate) or fired with the wrong cursor. + t.Errorf("unmet expectations (keyset pagination did not advance correctly): %v", err) + } +} + +// TestOrphanSweep_FetchLiveStackIDs_SecondPageError proves a DB error on a +// LATER keyset page (not just the first) propagates out — the loop must not +// silently return a partial set, which for PASS 5 could wrongly mark a live +// stack's namespace as an orphan. +func TestOrphanSweep_FetchLiveStackIDs_SecondPageError(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + + page1 := sqlmock.NewRows([]string{"id"}) + for i := 0; i < orphanLiveIDsBatchLimit; i++ { + page1.AddRow(fmt.Sprintf("stack-%06d", i)) + } + queryRE := `SELECT id::text\s+FROM stacks\s+WHERE id::text > \$1\s+ORDER BY id::text ASC\s+LIMIT \$2` + mock.ExpectQuery(queryRE).WillReturnRows(page1) + mock.ExpectQuery(queryRE).WillReturnError(errors.New("conn lost mid-sweep")) + + w := &OrphanSweepReconciler{db: db} + if _, err := w.fetchLiveStackIDs(context.Background()); err == nil { + t.Fatal("expected error from second-page query failure, got nil (partial set must NOT be returned)") + } +} + +// TestOrphanSweep_FetchLiveStackIDs_RowsErr proves a row-iteration error +// (rows.Err() non-nil — e.g. the connection drops mid-stream) propagates out +// rather than silently truncating the live-id set. Distinct from a +// QueryContext error: this fires AFTER rows start streaming. +func TestOrphanSweep_FetchLiveStackIDs_RowsErr(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + + rows := sqlmock.NewRows([]string{"id"}). + AddRow("stack-aaaa"). + RowError(0, errors.New("conn reset mid-stream")) + mock.ExpectQuery(`SELECT id::text\s+FROM stacks\s+WHERE id::text > \$1`). + WillReturnRows(rows) + + w := &OrphanSweepReconciler{db: db} + if _, err := w.fetchLiveStackIDs(context.Background()); err == nil { + t.Fatal("expected rows.Err() to propagate, got nil") + } +} + // TestOrphanSweep_BuildPass3Evidence_AbsentAndPresent covers both // branches of buildPass3Evidence (present + absent). func TestOrphanSweep_BuildPass3Evidence_AbsentAndPresent(t *testing.T) { diff --git a/internal/jobs/orphan_sweep_reconciler.go b/internal/jobs/orphan_sweep_reconciler.go index de6d1fd..76c1a0d 100644 --- a/internal/jobs/orphan_sweep_reconciler.go +++ b/internal/jobs/orphan_sweep_reconciler.go @@ -184,6 +184,16 @@ const ( // (a ghcr.io outage that wedged many builds at once) is drained over // several ticks rather than spamming the k8s API in one burst. orphanStuckBuildBatchLimit = 25 + + // orphanLiveIDsBatchLimit caps how many ids fetchLiveStackIDs pulls per + // round-trip. The full live-id set is still materialized into the + // returned map (PASS 5 needs the complete set to decide orphan-hood), + // but the rows are streamed in keyset-paginated batches rather than one + // unbounded SELECT — bounding the server-side cursor + per-fetch memory + // so a stacks table that grows to tens of thousands of rows cannot + // pin a multi-MB result set in one allocation. Keyset (id > $1 ORDER BY + // id) is restart-safe and index-friendly (PK scan, no OFFSET drift). + orphanLiveIDsBatchLimit = 1000 ) // customerNamespacePrefix is the prefix of every per-resource customer @@ -946,23 +956,61 @@ func (w *OrphanSweepReconciler) sweepOrphanedStackNamespaces(ctx context.Context // status — even a terminal-status stacks row pins its namespace so the // per-stack teardown path owns the delete. The pass is a strict "no row at // all = orphan" sweep. +// +// Batching (bug bash 2026-06-03): the previous `SELECT id::text FROM stacks` +// loaded the ENTIRE stacks table into one result set/allocation. This now +// streams the ids in keyset-paginated batches of orphanLiveIDsBatchLimit +// (WHERE id > $1 ORDER BY id LIMIT $2), so the server-side cursor + per-fetch +// memory stay bounded regardless of table size. The complete set is still +// returned — PASS 5 must see every live id to avoid deleting a live +// namespace — but it is assembled incrementally rather than in one shot. +// +// Keyset over OFFSET: an OFFSET sweep re-scans skipped rows each page and can +// skip/duplicate ids if rows are inserted/deleted mid-sweep; the (id > last) +// predicate rides the primary-key index and is stable under concurrent writes +// (a brand-new stack id either sorts after the cursor — seen this sweep — or +// before it — already seen; either way it lands in the set). Newly-inserted +// stacks during the sweep are the conservative case for PASS 5 anyway: a +// missed live id can only ever PRESERVE a namespace, never wrongly delete one. func (w *OrphanSweepReconciler) fetchLiveStackIDs(ctx context.Context) (map[string]bool, error) { - rows, err := w.db.QueryContext(ctx, `SELECT id::text FROM stacks`) - if err != nil { - return nil, err - } - defer func() { _ = rows.Close() }() out := make(map[string]bool) - for rows.Next() { - var id string - if scanErr := rows.Scan(&id); scanErr != nil { - return nil, scanErr + lastID := "" // keyset cursor: empty string sorts before every real id + for { + rows, err := w.db.QueryContext(ctx, ` + SELECT id::text + FROM stacks + WHERE id::text > $1 + ORDER BY id::text ASC + LIMIT $2 + `, lastID, orphanLiveIDsBatchLimit) + if err != nil { + return nil, err } - if id != "" { - out[id] = true + batchCount := 0 + for rows.Next() { + var id string + if scanErr := rows.Scan(&id); scanErr != nil { + _ = rows.Close() + return nil, scanErr + } + batchCount++ + lastID = id + if id != "" { + out[id] = true + } + } + if rowsErr := rows.Err(); rowsErr != nil { + _ = rows.Close() + return nil, rowsErr + } + _ = rows.Close() + // A short page (fewer rows than the limit) means we've drained the + // table — the last keyset query returned the tail. Stop. + if batchCount < orphanLiveIDsBatchLimit { + break } } - return out, rows.Err() + return out, nil } // ── PASS 6 — stuck-build detection (2026-05-20) ────────────────────────── diff --git a/internal/jobs/orphan_sweep_reconciler_test.go b/internal/jobs/orphan_sweep_reconciler_test.go index e0efdef..deb37e8 100644 --- a/internal/jobs/orphan_sweep_reconciler_test.go +++ b/internal/jobs/orphan_sweep_reconciler_test.go @@ -701,7 +701,8 @@ func TestOrphanSweep_Pass5_ReclaimsOrphanedStackNamespace(t *testing.T) { // PASS 4 (no customer namespaces — fake returns empty; short-circuits). // PASS 5: live-stack-ids query returns ONLY liveStackID → orphanNS is // the orphan. - mock.ExpectQuery(`SELECT id::text FROM stacks`). + mock.ExpectQuery(`SELECT id::text\s+FROM stacks\s+WHERE id::text > \$1\s+ORDER BY id::text ASC\s+LIMIT \$2`). + WithArgs("", orphanLiveIDsBatchLimit). WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(liveStackID)) lister := newFakeNamespaceLister().withStackNamespaces(liveNS, orphanNS)