fix(reaper+reconciler): MR-P0-1 + MR-P0-2 — stop the customer-namespace leak; wire real prober

BugBash 2026-05-20 P0s (cross-confirmed across 4 tracks — T1/T5/T20/T24).
The MR-P0-1 leak put 188 instant-customer-* namespaces, each running a live
Postgres/Redis/Mongo pod with no DB record, into the prod cluster.

MR-P0-1a (the headline finding): the reaper used to mark a resources row
status='deleted' EVEN WHEN provisioner.DeprovisionResource had returned an
error. A 'deleted' row is terminal and invisible to every reconciler, so the
backend (customer namespace + its live pod) was orphaned forever. Now the
reaper only advances the row to 'deleted' when the backend teardown
genuinely succeeded; on failure the row stays in its reapable status and the
next tick retries. A new instant_expire_deprovision_failed_total counter
surfaces a sustained leak rate to NR. Added a deprovisioner-seam interface
(ResourceDeprovisioner) so the regression test can inject a failing fake.

MR-P0-1b: the orphan_sweep_reconciler's PASS 3 only swept instant-deploy-*
namespaces — there was no sweep for instant-customer-*. New PASS 4 lists
every instant-customer-<token> namespace and deletes any whose token has no
active/paused/suspended resources row. Fail-open posture mirrors PASS 3 (RBAC
forbidden / DB blip → one WARN, zero orphans this sweep). This is the
durable backstop that stops the leak from recurring; 121 of the 188 leaked
namespaces had to be cleaned up by hand.

MR-P0-2: workers.go constructed the provisioner_reconciler with nil → the
fallback NoopProber whose every Probe returns ProbeReachable. The
reconciler would blindly promote stuck 'pending' rows to status='active'
WITHOUT checking the backend. Now wired with the same NewRealProber(cfg)
that resource_heartbeat already uses two lines below.

Regression tests (each fails without the fix, verified):

  P0-1a → TestExpireAnonymousWorker_P0_1a_DeprovisionFailure_DoesNotMarkDeleted
          (+ companion: ..._DeprovisionSuccess_StillMarksDeleted)
  P0-1b → TestOrphanSweep_Pass4_ReclaimsOrphanedCustomerNamespace
          (+ companion: TestOrphanSweep_Pass4_NoCustomerNamespaces_NoQuery)
  P0-2  → TestProvisionerReconciler_P0_2_RealProberUnreachable_DoesNotPromote
        + TestProvisionerReconciler_StartWorkersCallSite_PassesRealProber
          (call-site text-grep binding so a future nil revert fails CI)
        + TestProvisionerReconciler_WorkerKeepsRealProber

go build / go vet / go test ./... all green.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: mastermanas805

internal/jobs/expire.go

@@ -42,15 +42,28 @@ type ExpireAnonymousArgs struct{}
 
 func (ExpireAnonymousArgs) Kind() string { return "expire_anonymous" }
 
+// ResourceDeprovisioner is the narrow seam the reaper uses to tear down a
+// resource's physical backend (DROP DATABASE / DROP USER / delete NATS pod).
+// Lifted to an interface so a test can inject a fake that fails the
+// deprovision and assert the reaper does NOT then mark the row deleted
+// (MR-P0-1a, BugBash 2026-05-20). The concrete *provisioner.Client satisfies
+// it. nil = deprovision skipped (fail open) — same posture as before.
+type ResourceDeprovisioner interface {
+	DeprovisionResource(ctx context.Context, token, providerResourceID string, resType commonv1.ResourceType) error
+}
+
+// compile-time assertion: the real provisioner client satisfies the seam.
+var _ ResourceDeprovisioner = (*provisioner.Client)(nil)
+
 // ExpireAnonymousWorker expires anonymous resources that have passed their expires_at time.
 // It calls the provisioner to DROP the physical resource (DB/ACL user/Mongo user) before
 // marking the row as deleted, so credentials stop working immediately rather than lingering
 // until the next provisioner cycle.
 type ExpireAnonymousWorker struct {
 	river.WorkerDefaults[ExpireAnonymousArgs]
 	db          *sql.DB
-	provisioner *provisioner.Client // nil = deprovision skipped (fail open)
-	minioClient *madmin.AdminClient // nil = MinIO IAM-user cleanup skipped (legacy self-hosted MinIO backend)
+	provisioner ResourceDeprovisioner // nil = deprovision skipped (fail open)
+	minioClient *madmin.AdminClient   // nil = MinIO IAM-user cleanup skipped (legacy self-hosted MinIO backend)
 	// objectDeleter deletes a storage resource's objects under its tenant
 	// prefix on the S3-compatible OBJECT_STORE_* backend (DO Spaces in prod).
 	// This is the only path that actually removes a tenant's objects on
@@ -65,15 +78,34 @@ type ExpireAnonymousWorker struct {
 }
 
 // NewExpireAnonymousWorker constructs an ExpireAnonymousWorker.
-// Pass nil for provClient to skip physical deprovisioning (e.g. in tests or when the
-// provisioner is unavailable — the DB row is still marked deleted).
+// Pass nil for provClient to skip physical deprovisioning (e.g. in tests or
+// when the provisioner is unavailable — the deprovision step is then skipped,
+// and per MR-P0-1a the row is left in its reapable status for a later retry,
+// NOT marked deleted).
 // Pass nil for minioClient to skip MinIO IAM user cleanup.
 //
 // The storage-object deleter and bucket are wired separately via
 // WithObjectDeleter — callers that don't set it leave storage expiry as a
 // logged WARN (no silent no-op) rather than dropping the tenant's objects.
 func NewExpireAnonymousWorker(db *sql.DB, provClient *provisioner.Client, minioClient *madmin.AdminClient) *ExpireAnonymousWorker {
-	return &ExpireAnonymousWorker{db: db, provisioner: provClient, minioClient: minioClient}
+	w := &ExpireAnonymousWorker{db: db, minioClient: minioClient}
+	// A typed-nil *provisioner.Client stored straight into the interface
+	// field would make `w.provisioner != nil` true and panic on call. Only
+	// assign when the pointer is genuinely non-nil so the nil-skip guard in
+	// Work() behaves.
+	if provClient != nil {
+		w.provisioner = provClient
+	}
+	return w
+}
+
+// WithDeprovisioner overrides the deprovisioner seam — used by tests to
+// inject a fake that fails the deprovision call so the MR-P0-1a regression
+// (a failed deprovision must NOT mark the row deleted) can be exercised
+// without a live provisioner. Returns the worker for chaining.
+func (w *ExpireAnonymousWorker) WithDeprovisioner(d ResourceDeprovisioner) *ExpireAnonymousWorker {
+	w.provisioner = d
+	return w
 }
 
 // WithObjectDeleter wires the S3-compatible object deleter used to remove a
@@ -160,11 +192,26 @@ func (w *ExpireAnonymousWorker) Work(ctx context.Context, job *river.Job[ExpireA
 		return nil
 	}
 
-	// Step 2: For each candidate, deprovision the physical resource then mark the row deleted.
-	// Errors on either step are logged but never propagate — fail open so one bad resource
-	// does not block the expiry of the remaining batch.
+	// Step 2: For each candidate, deprovision the physical resource then mark
+	// the row deleted. One bad resource never blocks the expiry of the rest of
+	// the batch (fail open).
+	//
+	// MR-P0-1a (BugBash 2026-05-20, cross-confirmed by T1/T5/T20/T24): the row
+	// is marked status='deleted' ONLY when the backend teardown genuinely
+	// succeeded (or there was nothing to tear down). A 'deleted' row is
+	// terminal and invisible to every reconciler — so marking it deleted while
+	// the physical Postgres DB / Redis ACL / Mongo user / NATS pod is still
+	// live orphans that backend forever (it bills real money and consumes
+	// shared-cluster capacity). On a deprovision failure we instead leave the
+	// row in its current reapable status: the next reaper tick (or the team-
+	// deletion executor) retries the teardown.
 	var expired int
 	for _, r := range candidates {
+		// deprovisionFailed gates the mark-deleted UPDATE below. true = a
+		// backend teardown call returned an error this tick; the row stays
+		// reapable so a later tick retries it instead of stranding the infra.
+		deprovisionFailed := false
+
 		// Deprovision — credentials become invalid immediately.
 		switch r.resourceType {
 		case "storage":
@@ -199,18 +246,35 @@ func (w *ExpireAnonymousWorker) Work(ctx context.Context, job *river.Job[ExpireA
 					if deprovErr := w.provisioner.DeprovisionResource(
 						ctx, r.token, r.providerResourceID, resType,
 					); deprovErr != nil {
+						// MR-P0-1a: a failed teardown must NOT advance the
+						// row to 'deleted'. Flag it so the mark-deleted
+						// UPDATE below is skipped; the row stays reapable
+						// and the next tick retries the DROP.
+						deprovisionFailed = true
 						slog.Warn("jobs.expire_anonymous.deprovision_failed",
 							"error", deprovErr,
 							"resource_id", r.id,
 							"resource_type", r.resourceType,
 							"token", r.token,
 							"job_id", job.ID,
+							"effect", "row left reapable for retry; NOT marked deleted (MR-P0-1a)",
 						)
 					}
 				}
 			}
 		}
 
+		// MR-P0-1a: skip the mark-deleted UPDATE when the backend teardown
+		// failed this tick. Marking a row 'deleted' (a terminal, non-reapable
+		// status) with live backend infra behind it permanently orphans that
+		// infra — no reconciler ever revisits a 'deleted' row. Leaving the row
+		// reapable means the next reaper tick re-selects it and retries the
+		// DROP until it genuinely succeeds.
+		if deprovisionFailed {
+			metrics.ExpireDeprovisionFailedTotal.Inc()
+			continue
+		}
+
 		// Guarded UPDATE: mark deleted only from a non-terminal status, matching
 		// the SELECT above. Gating on status='active' alone would leave the
 		// paused/suspended rows we just deprovisioned stuck in their old status.

internal/jobs/expire_test.go

@@ -2,6 +2,8 @@ package jobs_test
 
 import (
 	"context"
+	"database/sql/driver"
+	"errors"
 	"sync"
 	"testing"
 
@@ -10,9 +12,35 @@ import (
 	"github.com/riverqueue/river"
 	"github.com/riverqueue/river/rivertype"
 
+	commonv1 "instant.dev/proto/common/v1"
 	"instant.dev/worker/internal/jobs"
 )
 
+// recordingArg is a sqlmock.Argument matcher that records whether it was ever
+// evaluated. It matches ANY value — its only job is to observe that the
+// statement it gates was actually executed. Used by the MR-P0-1a regression
+// test to detect that the reaper attempted the mark-deleted UPDATE.
+type recordingArg struct{ hit bool }
+
+func (r *recordingArg) Match(_ driver.Value) bool {
+	r.hit = true
+	return true
+}
+
+// fakeDeprovisioner is a jobs.ResourceDeprovisioner test double. It records
+// every DeprovisionResource call and can be told to fail — used by the
+// MR-P0-1a regression test to assert the reaper does NOT mark a row deleted
+// when the backend teardown errors.
+type fakeDeprovisioner struct {
+	calls   int
+	failErr error // non-nil → every DeprovisionResource call fails
+}
+
+func (f *fakeDeprovisioner) DeprovisionResource(_ context.Context, _, _ string, _ commonv1.ResourceType) error {
+	f.calls++
+	return f.failErr
+}
+
 // fakeObjectDeleter is a fake S3BackupDeleter used to assert the storage-
 // expiry path actually drives an object delete against the object store.
 // It records the ListObjects prefixes it was asked for and emits the
@@ -293,3 +321,107 @@ func TestExpireAnonymousWorker_StorageExpiry_NoDeleterWarns(t *testing.T) {
 		t.Errorf("unmet expectations: %v", err)
 	}
 }
+
+// TestExpireAnonymousWorker_P0_1a_DeprovisionFailure_DoesNotMarkDeleted is the
+// MR-P0-1a regression guard (BugBash 2026-05-20, cross-confirmed by
+// T1/T5/T20/T24 — the headline namespace/resource leak).
+//
+// THE BUG: the reaper called provisioner.DeprovisionResource, and on an error
+// logged a WARN but FELL THROUGH to the guarded `UPDATE resources SET
+// status='deleted'`. A 'deleted' row is terminal and invisible to every
+// reconciler — so the backend (the customer's instant-customer-<token> k8s
+// namespace and its live Postgres/Redis/Mongo pod) was orphaned forever,
+// billing real money. 188 such namespaces leaked in prod.
+//
+// THE FIX: on a deprovision error the row is LEFT in its reapable status; the
+// next reaper tick retries the teardown. The row is marked 'deleted' only
+// after a genuinely successful backend teardown.
+//
+// THE ASSERTION: with a deprovisioner that always fails, the worker must
+// issue NO `UPDATE resources SET status='deleted'`. sqlmock has ordered
+// expectations — only the SELECT and the trailing active-anon COUNT are
+// queued; an unexpected UPDATE makes ExpectationsWereMet fail. If a future
+// edit reintroduces the unconditional mark-deleted, this test fails.
+func TestExpireAnonymousWorker_P0_1a_DeprovisionFailure_DoesNotMarkDeleted(t *testing.T) {
+	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
+	if err != nil {
+		t.Fatalf("sqlmock.New: %v", err)
+	}
+	defer db.Close()
+
+	// One expired postgres resource — has a real provider_resource_id so the
+	// reaper attempts a deprovision RPC.
+	mock.ExpectQuery(`SELECT id::text`).
+		WillReturnRows(sqlmock.NewRows([]string{"id", "token", "resource_type", "provider_resource_id"}).
+			AddRow("id-leak", "tok-leak", "postgres", "db_tok_leak"))
+	// The mark-deleted UPDATE IS queued — but gated by markDeletedSpy, a
+	// recording argument matcher. With the fix the reaper SKIPS this UPDATE
+	// (deprovision failed), so the matcher never fires and markDeletedSpy.hit
+	// stays false. With the bug the reaper FALLS THROUGH to this UPDATE, the
+	// matcher fires, and hit flips true — failing the test. Queuing it (rather
+	// than omitting it) is what makes the buggy path observable: an omitted
+	// expectation just produces a swallowed sqlmock error inside the reaper's
+	// fail-open `continue`, which ExpectationsWereMet does NOT surface.
+	markDeletedSpy := &recordingArg{}
+	mock.ExpectExec(`UPDATE resources SET status = 'deleted'`).
+		WithArgs(markDeletedSpy).
+		WillReturnResult(sqlmock.NewResult(1, 1))
+	// The trailing active-anon count always runs.
+	mock.ExpectQuery(`SELECT COUNT\(\*\) FROM resources`).
+		WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0))
+
+	deprov := &fakeDeprovisioner{failErr: errors.New("provisioner unreachable: context deadline exceeded")}
+	w := jobs.NewExpireAnonymousWorker(db, nil, nil).WithDeprovisioner(deprov)
+
+	if err := w.Work(context.Background(), fakeJob[jobs.ExpireAnonymousArgs]()); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	// The reaper must have actually attempted the teardown.
+	if deprov.calls != 1 {
+		t.Errorf("DeprovisionResource calls = %d, want 1 (the reaper must attempt teardown)", deprov.calls)
+	}
+	// THE LOAD-BEARING ASSERTION: the mark-deleted UPDATE must NOT have run.
+	// markDeletedSpy.hit is true iff the reaper executed `UPDATE ... SET
+	// status='deleted'` for the row whose deprovision failed — the exact bug
+	// that orphans the customer's instant-customer-<token> namespace forever.
+	if markDeletedSpy.hit {
+		t.Error("MR-P0-1a regression: the reaper marked the row status='deleted' " +
+			"even though DeprovisionResource FAILED. A 'deleted' row is terminal and " +
+			"invisible to every reconciler — the backend (customer k8s namespace + its " +
+			"live DB/Redis pod) is now orphaned forever. On a failed deprovision the row " +
+			"must stay reapable so the next tick retries.")
+	}
+}
+
+// TestExpireAnonymousWorker_P0_1a_DeprovisionSuccess_StillMarksDeleted is the
+// companion to the above: it pins that the fix did NOT break the happy path —
+// when the backend teardown genuinely succeeds, the row IS marked 'deleted'.
+func TestExpireAnonymousWorker_P0_1a_DeprovisionSuccess_StillMarksDeleted(t *testing.T) {
+	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
+	if err != nil {
+		t.Fatalf("sqlmock.New: %v", err)
+	}
+	defer db.Close()
+
+	mock.ExpectQuery(`SELECT id::text`).
+		WillReturnRows(sqlmock.NewRows([]string{"id", "token", "resource_type", "provider_resource_id"}).
+			AddRow("id-ok", "tok-ok", "postgres", "db_tok_ok"))
+	// Deprovision succeeds → the mark-deleted UPDATE MUST run.
+	mock.ExpectExec(`UPDATE resources SET status = 'deleted'`).
+		WillReturnResult(sqlmock.NewResult(1, 1))
+	mock.ExpectQuery(`SELECT COUNT\(\*\) FROM resources`).
+		WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0))
+
+	deprov := &fakeDeprovisioner{} // failErr nil → succeeds
+	w := jobs.NewExpireAnonymousWorker(db, nil, nil).WithDeprovisioner(deprov)
+
+	if err := w.Work(context.Background(), fakeJob[jobs.ExpireAnonymousArgs]()); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if deprov.calls != 1 {
+		t.Errorf("DeprovisionResource calls = %d, want 1", deprov.calls)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("happy path regressed: a successful deprovision must still mark the row deleted: %v", err)
+	}
+}

internal/jobs/k8s_namespace_client.go

@@ -95,15 +95,31 @@ func (c *k8sNamespaceClient) NamespaceExists(ctx context.Context, namespace stri
 // on any API failure so the reconciler skips the k8s-orphan phase rather
 // than acting on a truncated list.
 func (c *k8sNamespaceClient) ListDeployNamespaces(ctx context.Context) ([]string, error) {
+	return c.listNamespacesWithPrefix(ctx, deployNamespacePrefixTDE)
+}
+
+// ListCustomerNamespaces returns the names of every instant-customer-*
+// namespace currently in the cluster. The orphan-sweep reconciler's PASS 4
+// uses this to find customer namespaces whose backing resources row is gone
+// (the MR-P0-1b leak: a reaper that marked the row 'deleted' while the
+// namespace's backend stayed live). Returns an error on any API failure so
+// the reconciler skips the customer-orphan phase rather than acting on a
+// truncated list — never delete a namespace off an incomplete picture.
+func (c *k8sNamespaceClient) ListCustomerNamespaces(ctx context.Context) ([]string, error) {
+	return c.listNamespacesWithPrefix(ctx, customerNamespacePrefix)
+}
+
+// listNamespacesWithPrefix is the shared cluster-scoped namespace List used
+// by both ListDeployNamespaces and ListCustomerNamespaces.
+func (c *k8sNamespaceClient) listNamespacesWithPrefix(ctx context.Context, prefix string) ([]string, error) {
 	list, err := c.cs.CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
 	if err != nil {
-		return nil, fmt.Errorf("k8sNamespaceClient.ListDeployNamespaces: %w", err)
+		return nil, fmt.Errorf("k8sNamespaceClient.listNamespacesWithPrefix %q: %w", prefix, err)
 	}
 	var out []string
 	for i := range list.Items {
 		name := list.Items[i].Name
-		if len(name) >= len(deployNamespacePrefixTDE) &&
-			name[:len(deployNamespacePrefixTDE)] == deployNamespacePrefixTDE {
+		if len(name) >= len(prefix) && name[:len(prefix)] == prefix {
 			out = append(out, name)
 		}
 	}