Verify CP pod existence during startup orphan sweep

When a CP pod is force-deleted (crash, preemption, node drain), the
cp_instances row stays in 'active' state until the janitor expires the
heartbeat (~20s) and the orphan grace period passes (~30s). If the
deployment recreates the CP pod before that, the startup sweep sees the
dead CP as "live" in the DB and leaves its orphaned worker pods alone.

This caused TestK8sCPDeletionGarbageCollects to fail: the test force-
deletes a CP pod and expects its workers to be cleaned up within 90s,
but the startup sweep was deferring to the janitor path which is slower.

The fix: for each "live" CP returned by ListLiveControlPlaneInstanceIDs,
extract the pod name from the instance ID (format "pod-name:boot-id")
and verify the pod actually exists in K8s. If it's gone, treat the CP
as dead for the purpose of the sweep. This is ground truth — the DB
may lag behind K8s, but K8s is authoritative on pod existence.

Self is always skipped (the current CP is explicitly added to the live
set afterward), so this only checks peers.

Also fixes the existing peer-preservation tests to create CP pods for
peers that should be considered alive, now that the code verifies pod
existence.

Author: benben

controlplane/k8s_pool.go

@@ -233,6 +233,19 @@ func (p *K8sWorkerPool) cleanupOrphanedWorkers() {
 			return
 		}
 		for _, id := range ids {
+			// The DB may lag behind K8s: a CP marked "active" in the DB
+			// could have been force-deleted (crash, preemption) and the
+			// janitor hasn't expired it yet. Verify the CP's pod actually
+			// exists before trusting the DB. The instance ID format is
+			// "pod-name:boot-id" — extract the pod name and check K8s.
+			podName := strings.SplitN(id, ":", 2)[0]
+			if podName != "" && id != p.cpInstanceID && podName != p.cpID {
+				_, getErr := p.clientset.CoreV1().Pods(p.namespace).Get(ctx, podName, metav1.GetOptions{})
+				if errors.IsNotFound(getErr) {
+					slog.Info("Live CP in DB but pod is gone; treating as dead for orphan sweep.", "cp", id, "pod", podName)
+					continue
+				}
+			}
 			liveCPIDs[id] = true
 			liveCPLabels[controlPlaneIDLabelValue(id)] = true
 		}

controlplane/k8s_pool_test.go

@@ -1958,6 +1958,61 @@ func TestCleanupOrphanedWorkers_DeletesPodsFromDeadCPs(t *testing.T) {
 	}
 }
 
+func TestCleanupOrphanedWorkers_DeletesPodsFromCrashedLivePeer(t *testing.T) {
+	// A CP that is still "active" in the config store but whose pod was
+	// force-deleted (crash, preemption) should be treated as dead by the
+	// startup sweep. Without this, the worker pods from the crashed CP
+	// would linger until the janitor expires the heartbeat (~50s), failing
+	// the TestK8sCPDeletionGarbageCollects integration test.
+	pool, cs := newTestK8sPool(t, 5)
+	pool.cpInstanceID = "cp-self:boot-123"
+	crashedPeerID := "cp-crashed:boot-456"
+
+	store := &captureRuntimeWorkerStore{
+		// DB says cp-crashed is live (janitor hasn't expired it yet)
+		liveCPIDs: []string{"cp-self:boot-123", crashedPeerID},
+	}
+	pool.runtimeStore = store
+
+	// Create the self CP pod (must exist) but NOT the crashed peer's pod.
+	_, err := cs.CoreV1().Pods("default").Create(context.Background(), &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "cp-self",
+			Namespace: "default",
+		},
+	}, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Note: no "cp-crashed" pod created — simulates force-delete.
+
+	// Worker pod from the crashed peer.
+	_, err = cs.CoreV1().Pods("default").Create(context.Background(), &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "duckgres-worker-crashed-peer",
+			Namespace: "default",
+			Labels: map[string]string{
+				"app":                     "duckgres-worker",
+				"duckgres/cp-instance-id": controlPlaneIDLabelValue(crashedPeerID),
+			},
+		},
+	}, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pool.cleanupOrphanedWorkers()
+
+	pods, _ := cs.CoreV1().Pods("default").List(context.Background(), metav1.ListOptions{
+		LabelSelector: "app=duckgres-worker",
+	})
+	for _, p := range pods.Items {
+		if p.Name == "duckgres-worker-crashed-peer" {
+			t.Error("expected crashed peer's worker pod to be deleted — CP pod is gone even though DB says 'active'")
+		}
+	}
+}
+
 func TestCleanupOrphanedWorkers_PreservesLivePeerPods(t *testing.T) {
 	pool, cs := newTestK8sPool(t, 5)
 	pool.cpInstanceID = "cp-self:boot-123"
@@ -1969,6 +2024,11 @@ func TestCleanupOrphanedWorkers_PreservesLivePeerPods(t *testing.T) {
 	}
 	pool.runtimeStore = store
 
+	// Create the live peer's CP pod so the pod-existence check passes.
+	_, _ = cs.CoreV1().Pods("default").Create(context.Background(), &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: "cp-peer", Namespace: "default"},
+	}, metav1.CreateOptions{})
+
 	for _, tc := range []struct {
 		name       string
 		instanceID string
@@ -2093,10 +2153,15 @@ func TestCleanupOrphanedWorkers_MarksStaleIdleRowsLost(t *testing.T) {
 }
 
 func TestCleanupOrphanedWorkers_PreservesInflightSpawnsByLivePeers(t *testing.T) {
-	pool, _ := newTestK8sPool(t, 5)
+	pool, cs := newTestK8sPool(t, 5)
 	pool.cpInstanceID = "cp-self:boot-123"
 	peerID := "cp-peer:boot-456"
 
+	// Create the live peer's CP pod so the pod-existence check passes.
+	_, _ = cs.CoreV1().Pods("default").Create(context.Background(), &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: "cp-peer", Namespace: "default"},
+	}, metav1.CreateOptions{})
+
 	stale := time.Now().Add(-2 * time.Hour)
 	store := &captureRuntimeWorkerStore{
 		liveCPIDs: []string{"cp-self:boot-123", peerID},
@@ -2143,6 +2208,11 @@ func TestCleanupOrphanedWorkers_PreservesDrainingPeerWorkers(t *testing.T) {
 	pool.cpInstanceID = "cp-self:boot-123"
 	drainingPeerID := "cp-draining-peer:boot-456"
 
+	// Create the draining peer's CP pod — it's still running (draining, not dead).
+	_, _ = cs.CoreV1().Pods("default").Create(context.Background(), &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: "cp-draining-peer", Namespace: "default"},
+	}, metav1.CreateOptions{})
+
 	// The mock returns whatever the test sets in liveCPIDs; the production
 	// store query is `state <> expired` so callers always get both active
 	// and draining CPs in this list.