Skip retired/lost workers in janitor orphan list (#405)

The janitor's ListOrphanedWorkers query included workers in the retired
and lost states alongside active states. When a worker's owning CP is
expired, retireClaimedWorker re-persists the row with the original
expired CP as owner, so the orphan query keeps re-listing the same row
on every cycle. retireWorkerPod then loops on a 404 from the K8s pod
delete:

  Retiring K8s worker. id=685 pod=duckgres-worker-...-685
  Failed to delete worker pod. error="pods ... not found"

The original reason for including retired/lost was to let a successor
janitor finish a K8s pod delete that the dying CP didn't complete. That
case is now handled authoritatively by the K8s label-based startup scan
in cleanupOrphanedWorkerPods (4e96d11), so the inclusion is redundant
and harmful.

Excludes WorkerStateRetired and WorkerStateLost from the cleanup states.
First-time orphans still get processed: their state is reserved/idle/
spawning/etc. when the janitor first sees them, transitions to retired
inside retireClaimedWorker, then the next cycle skips the row.

Updates TestListOrphanedAndStuckWorkersPostgres to assert the new
behavior with both retired and lost orphan rows present.

Author: benben

controlplane/configstore/store.go

@@ -784,9 +784,11 @@ func (cs *ConfigStore) CreateNeutralWarmWorkerSlot(ownerCPInstanceID, podNamePre
 
 // ListOrphanedWorkers returns workers whose owning control-plane instance has
 // already been marked expired long enough ago to pass the orphan grace cutoff.
-// Retired/lost rows are included so a replacement janitor can finish deleting
-// worker pods when the original control plane died after persisting retirement
-// but before the Kubernetes delete completed.
+// Retired/lost rows are deliberately excluded: their pods are either already
+// gone (in which case re-listing the row would loop the janitor on a 404 from
+// the K8s delete forever) or were leaked when the previous CP died mid-delete,
+// and that leak case is handled authoritatively by the K8s label-based startup
+// scan in K8sWorkerPool.cleanupOrphanedWorkerPods.
 func (cs *ConfigStore) ListOrphanedWorkers(before time.Time) ([]WorkerRecord, error) {
 	var workers []WorkerRecord
 	cleanupStates := []WorkerState{
@@ -797,8 +799,6 @@ func (cs *ConfigStore) ListOrphanedWorkers(before time.Time) ([]WorkerRecord, er
 		WorkerStateHot,
 		WorkerStateHotIdle,
 		WorkerStateDraining,
-		WorkerStateRetired,
-		WorkerStateLost,
 	}
 	workerTable := cs.runtimeTable((&WorkerRecord{}).TableName())
 	cpTable := cs.runtimeTable((&ControlPlaneInstance{}).TableName())

tests/configstore/runtime_store_postgres_test.go

@@ -476,6 +476,10 @@ func TestListOrphanedAndStuckWorkersPostgres(t *testing.T) {
 	}); err != nil {
 		t.Fatalf("UpsertWorkerRecord(orphaned): %v", err)
 	}
+	// Retired and lost rows whose owning CP is expired must NOT be returned:
+	// their pods are already deleted (or will be cleaned up by the K8s label
+	// scan on the next CP startup), so re-listing them would loop the janitor
+	// on a 404 from the K8s pod delete forever.
 	if err := store.UpsertWorkerRecord(&configstore.WorkerRecord{
 		WorkerID:          63,
 		PodName:           "duckgres-worker-63",
@@ -488,6 +492,18 @@ func TestListOrphanedAndStuckWorkersPostgres(t *testing.T) {
 	}); err != nil {
 		t.Fatalf("UpsertWorkerRecord(retired orphan): %v", err)
 	}
+	if err := store.UpsertWorkerRecord(&configstore.WorkerRecord{
+		WorkerID:          64,
+		PodName:           "duckgres-worker-64",
+		State:             configstore.WorkerStateLost,
+		OrgID:             "analytics",
+		OwnerCPInstanceID: "cp-expired:boot-a",
+		OwnerEpoch:        4,
+		LastHeartbeatAt:   now.Add(-time.Minute),
+		RetireReason:      "crash",
+	}); err != nil {
+		t.Fatalf("UpsertWorkerRecord(lost orphan): %v", err)
+	}
 	if err := store.UpsertWorkerRecord(&configstore.WorkerRecord{
 		WorkerID:          62,
 		PodName:           "duckgres-worker-62",
@@ -509,11 +525,8 @@ func TestListOrphanedAndStuckWorkersPostgres(t *testing.T) {
 	if err != nil {
 		t.Fatalf("ListOrphanedWorkers: %v", err)
 	}
-	if len(orphaned) != 2 {
-		t.Fatalf("expected orphaned worker 61, got %#v", orphaned)
-	}
-	if orphaned[0].WorkerID != 61 || orphaned[1].WorkerID != 63 {
-		t.Fatalf("expected orphaned workers 61 and 63, got %#v", orphaned)
+	if len(orphaned) != 1 || orphaned[0].WorkerID != 61 {
+		t.Fatalf("expected only active orphaned worker 61, got %#v", orphaned)
 	}
 
 	stuck, err := store.ListStuckWorkers(now.Add(-2*time.Minute), now.Add(-2*time.Minute))