Retry duplicate runtime worker claims

controlplane/k8s_pool.go

@@ -1578,16 +1578,9 @@ func (p *K8sWorkerPool) reserveClaimedWorker(ctx context.Context, claimed *confi
 		p.mu.Unlock()
 		return nil, fmt.Errorf("worker %d claimed with stale owner epoch %d behind current %d: %w", claimed.WorkerID, claimed.OwnerEpoch, currentEpoch, errStaleRuntimeWorkerClaim)
 	}
-	if isSameRuntimeWorkerClaim(worker, claimed, assignment) {
-		worker.reservedAt = time.Now()
-		observeWarmPoolLifecycleGauges(p.workers)
+	if isDuplicateRuntimeWorkerClaim(worker, claimed, assignment) {
 		p.mu.Unlock()
-		if err := p.checkReservedWorkerLiveness(ctx, worker); err != nil {
-			slog.Warn("Claimed worker failed liveness recheck.", "worker", worker.ID, "worker_pod", worker.PodName(), "error", err)
-			p.retireWorkerWithReason(worker.ID, RetireReasonCrash)
-			return nil, err
-		}
-		return worker, nil
+		return nil, fmt.Errorf("worker %d already has in-flight claim for owner epoch %d: %w", claimed.WorkerID, claimed.OwnerEpoch, errStaleRuntimeWorkerClaim)
 	}
 	worker.SetOwnerCPInstanceID(claimed.OwnerCPInstanceID)
 	worker.SetOwnerEpoch(claimed.OwnerEpoch)
@@ -1617,7 +1610,7 @@ func (p *K8sWorkerPool) reserveClaimedWorker(ctx context.Context, claimed *confi
 	return worker, nil
 }
 
-func isSameRuntimeWorkerClaim(worker *ManagedWorker, claimed *configstore.WorkerRecord, assignment *WorkerAssignment) bool {
+func isDuplicateRuntimeWorkerClaim(worker *ManagedWorker, claimed *configstore.WorkerRecord, assignment *WorkerAssignment) bool {
 	if worker == nil || claimed == nil || assignment == nil {
 		return false
 	}

controlplane/k8s_pool_test.go

@@ -1358,7 +1358,7 @@ func TestK8sPoolReserveSharedWorkerClaimsRuntimeWorkerAndAdoptsPod(t *testing.T)
 	}
 }
 
-func TestK8sPoolReserveClaimedWorkerIsIdempotentForSameActivatingClaim(t *testing.T) {
+func TestK8sPoolReserveClaimedWorkerRejectsDuplicateActivatingClaim(t *testing.T) {
 	pool, _ := newTestK8sPool(t, 5)
 	assignment := &WorkerAssignment{OrgID: "analytics"}
 	worker := &ManagedWorker{ID: 44, podName: "duckgres-worker-test-cp-44", done: make(chan struct{})}
@@ -1371,32 +1371,23 @@ func TestK8sPoolReserveClaimedWorkerIsIdempotentForSameActivatingClaim(t *testin
 		t.Fatalf("SetSharedState: %v", err)
 	}
 	pool.workers[worker.ID] = worker
-	pool.healthCheckFunc = func(ctx context.Context, got *ManagedWorker) error {
-		if got.ID != worker.ID {
-			t.Fatalf("expected liveness check for worker %d, got %d", worker.ID, got.ID)
-		}
-		return nil
-	}
 
-	got, err := pool.reserveClaimedWorker(context.Background(), &configstore.WorkerRecord{
+	_, err := pool.reserveClaimedWorker(context.Background(), &configstore.WorkerRecord{
 		WorkerID:          worker.ID,
 		PodName:           worker.PodName(),
 		State:             configstore.WorkerStateReserved,
 		OrgID:             assignment.OrgID,
 		OwnerCPInstanceID: pool.cpInstanceID,
 		OwnerEpoch:        5,
 	}, assignment)
-	if err != nil {
-		t.Fatalf("reserveClaimedWorker: %v", err)
-	}
-	if got != worker {
-		t.Fatalf("expected same worker instance")
+	if !errors.Is(err, errStaleRuntimeWorkerClaim) {
+		t.Fatalf("expected stale claim error, got %v", err)
 	}
-	if got.SharedState().Lifecycle != WorkerLifecycleActivating {
-		t.Fatalf("expected lifecycle to remain activating, got %q", got.SharedState().Lifecycle)
+	if worker.SharedState().Lifecycle != WorkerLifecycleActivating {
+		t.Fatalf("expected lifecycle to remain activating, got %q", worker.SharedState().Lifecycle)
 	}
-	if got.OwnerEpoch() != 5 {
-		t.Fatalf("expected owner epoch 5, got %d", got.OwnerEpoch())
+	if worker.OwnerEpoch() != 5 {
+		t.Fatalf("expected owner epoch 5, got %d", worker.OwnerEpoch())
 	}
 }