cache: recover from PVC-informer race in addPod

A Pod ADD event can reach the scheduler before its referenced PVC ADD
reaches the PVC informer — for example when a Pod is created together
with an OnDemand PVC in quick succession (Spark's k8s scheduler-backend
does exactly this for executor pods with `claimName=OnDemand`).

Since v1.15.0, the resulting `looking up PVC ... not found` error from
NewTaskInfo permanently strands the pod:

    addPod → NewTaskInfo → addPodCSIVolumesToTask
      → pvcInformer.Lister().Get(pvcName) returns not-found
      → klog.Errorf(...)
      → sc.resyncTask(pi)        // enqueues taskKey into errTasks
      → return err               // task never enters sc.Jobs

    processResyncTask → errTasks.Get() → parseErrTaskKey → getTaskByUID
      → "failed to find task <uid>"
      → errTasks.Forget(taskKey) // permanent
      → return                   // never reaches retryResyncTask

There is no PVC-informer AddFunc that re-queues stuck pods, and the
pod's resourceVersion never advances before scheduling, so
`UpdatePod → deletePod + addPod` never fires. Executor pods stay
Pending; the PVC stays Pending with `WaitForPodScheduled`.

Before volcano-sh#5058, `addPod`'s error branch fell through to `sc.addTask(pi)`
even after `NewTaskInfo` returned an error, so the incomplete task
lived in `sc.Jobs` and processResyncTask could find it by UID on the
next tick and call syncTask, which refetches the pod from the apiserver
— by which point the PVC informer has synced. volcano-sh#5058 added a DRA-specific
happy path using the correct `addTask + resyncTask + return nil`
pattern, but also added a generic `return err` that removed the
implicit recovery for PVC-not-found.

Extend the DRA branch to PVC-not-found errors, using the same pattern.
Change getPodCSIVolumes to wrap the lister error with %w so callers
can classify it with errors.Is / apierrors.IsNotFound instead of
matching on the message string.

Adds `TestAddPodWithUnresolvedPVCCachesTaskForResync` mirroring the
existing DRA counterpart.

Version scope: only v1.15.0 is affected. Verified with
`git merge-base --is-ancestor 5ba79f3 <tag>` — every v1.14.x tag
through v1.14.3 (released after v1.15.0) is clean; v1.15.0-alpha.0 is
also clean.

Signed-off-by: Gustavo Parcianello <gustavo.parcianello@sap.com>

Author: gucardona

pkg/scheduler/cache/cache_mock.go

@@ -164,6 +164,13 @@ func (sc *SchedulerCache) initMockInformers() {
 	// Create node informer
 	sc.nodeInformer = informerFactory.Core().V1().Nodes()
 
+	// Create PVC informer (required by addPodCSIVolumesToTask for pods
+	// referencing a PersistentVolumeClaim). Materialise via Informer() so
+	// SharedInformerFactory.Start actually starts it — matches the real
+	// newSchedulerCache wiring.
+	sc.pvcInformer = informerFactory.Core().V1().PersistentVolumeClaims()
+	sc.pvcInformer.Informer()
+
 	// Initialize DRA manager if feature is enabled
 	if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.DynamicResourceAllocation) {
 		ctx := context.TODO()

pkg/scheduler/cache/event_handlers.go

@@ -129,7 +129,12 @@ func (sc *SchedulerCache) getPodCSIVolumes(pod *v1.Pod) (map[v1.ResourceName]int
 			// The PVC is required to proceed with
 			// scheduling of a new pod because it cannot
 			// run without it. Bail out immediately.
-			return volumes, fmt.Errorf("looking up PVC %s/%s: %v", pod.Namespace, pvcName, err)
+			//
+			// Wrap with %w so callers can distinguish
+			// "PVC not yet in informer" (not-found) from
+			// other lookup failures via errors.Is /
+			// apierrors.IsNotFound and recover accordingly.
+			return volumes, fmt.Errorf("looking up PVC %s/%s: %w", pod.Namespace, pvcName, err)
 		}
 		// The PVC for an ephemeral volume must be owned by the pod.
 		if isEphemeral {
@@ -271,6 +276,16 @@ func (sc *SchedulerCache) addPod(pod *v1.Pod) error {
 			sc.resyncTask(pi)
 			return nil
 		}
+		// Recover from a Pod ADD that races ahead of its PVC ADD on
+		// the informers. Same pattern as the DRA branch above.
+		if errors.IsNotFound(err) {
+			klog.V(4).Infof("PVC for pod <%s/%s> not yet in informer cache, add task and retry: %v", pod.Namespace, pod.Name, err)
+			if addErr := sc.addTask(pi); addErr != nil {
+				return addErr
+			}
+			sc.resyncTask(pi)
+			return nil
+		}
 		klog.Errorf("generate taskInfo for pod(%s) failed: %v", pod.Name, err)
 		sc.resyncTask(pi)
 		return err

pkg/scheduler/cache/event_handlers_pvc_race_test.go

@@ -0,0 +1,121 @@
+/*
+Copyright 2026 The Volcano Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cache
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	schedulingv1beta1 "volcano.sh/apis/pkg/apis/scheduling/v1beta1"
+	schedulingapi "volcano.sh/volcano/pkg/scheduler/api"
+)
+
+// TestAddPodWithUnresolvedPVCCachesTaskForResync verifies the full
+// recovery path when a Pod ADD races ahead of its PVC ADD:
+//
+//  1. addPod caches the task and enqueues it for resync instead of
+//     permanently dropping the pod.
+//  2. Once the PVC arrives on the informer, one processResyncTask
+//     tick refreshes the task via syncTask and removes it from
+//     errTasks — the pod is now visible to the scheduler.
+//
+// Mirrors the DRA counterpart TestAddPodWithUnresolvedResourceClaimTemplateCachesTaskForResync,
+// extended with the end-to-end recovery assertion.
+func TestAddPodWithUnresolvedPVCCachesTaskForResync(t *testing.T) {
+	sc := newMockSchedulerCache("volcano")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	pod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "pod-with-pvc",
+			Namespace: "default",
+			UID:       types.UID("pod-with-pvc-uid"),
+			Annotations: map[string]string{
+				schedulingv1beta1.KubeGroupNameAnnotationKey: "pg-with-pvc",
+			},
+		},
+		Spec: v1.PodSpec{
+			SchedulerName: "volcano",
+			Volumes: []v1.Volume{
+				{
+					Name: "scratch",
+					VolumeSource: v1.VolumeSource{
+						PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+							ClaimName: "eventual-pvc",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	// Seed the pod in the fake kube-client so syncTask's later
+	// Pods(ns).Get(name) call finds it on the retry tick.
+	_, err := sc.kubeClient.CoreV1().Pods(pod.Namespace).Create(ctx, pod, metav1.CreateOptions{})
+	assert.NoError(t, err)
+
+	sc.informerFactory.Start(ctx.Done())
+	for informer, synced := range sc.informerFactory.WaitForCacheSync(ctx.Done()) {
+		assert.Truef(t, synced, "informer %v failed to sync", informer)
+	}
+
+	// Phase 1: PVC ADD has not arrived yet. addPod must defer to resync.
+	err = sc.addPod(pod)
+	assert.NoError(t, err, "addPod must defer to the retry loop when the PVC is missing from the informer cache")
+
+	job, found := sc.Jobs[schedulingapi.JobID("default/pg-with-pvc")]
+	assert.True(t, found, "job must be added to sc.Jobs so processResyncTask can find its task by UID")
+	if assert.NotNil(t, job) {
+		_, found = job.Tasks[schedulingapi.TaskID("pod-with-pvc-uid")]
+		assert.True(t, found, "task must be added to job.Tasks so getTaskByUID succeeds on the retry tick")
+	}
+	assert.Equal(t, 1, sc.errTasks.Len(), "one entry must be enqueued on errTasks so processResyncTask visits this task")
+
+	// Phase 2: PVC arrives on the informer. processResyncTask should
+	// refresh the task via syncTask and remove it from errTasks.
+	pvc := &v1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "eventual-pvc",
+			Namespace: "default",
+		},
+	}
+	_, err = sc.kubeClient.CoreV1().PersistentVolumeClaims(pvc.Namespace).Create(ctx, pvc, metav1.CreateOptions{})
+	assert.NoError(t, err)
+
+	// Wait for the PVC informer to observe the new PVC.
+	err = wait.PollUntilContextTimeout(ctx, 20*time.Millisecond, 2*time.Second, true, func(context.Context) (bool, error) {
+		_, err := sc.pvcInformer.Lister().PersistentVolumeClaims("default").Get("eventual-pvc")
+		return err == nil, nil
+	})
+	assert.NoError(t, err, "PVC never appeared on the informer lister")
+
+	// Drive one iteration of the resync worker. This pops the taskKey,
+	// calls syncTask (fresh kubeClient Get → NewTaskInfo → PVC lookup
+	// now succeeds), and Forgets the taskKey on success.
+	sc.processResyncTask()
+
+	assert.Equal(t, 0, sc.errTasks.Len(), "task must be removed from errTasks once the PVC has synced and syncTask succeeds")
+}