Fix race condition: check concurrent limit before GetExposed

Add CanAcceptNewTask() method to datapath.Manager to check if a new
task can be accepted before executing expensive GetExposed operation.

This prevents multiple tasks from wasting time on GetExposed when
the concurrent limit is already reached, which was causing tasks
to get ConcurrentLimitExceed errors after spending time on GetExposed.

Applied to all controllers:
- DataUploadReconciler
- DataDownloadReconciler
- PodVolumeBackupReconciler
- PodVolumeRestoreReconciler

Signed-off-by: Andrei Kvapil <[email protected]>

Author: kvaps

pkg/controller/data_download_controller.go

@@ -315,6 +315,14 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 			return ctrl.Result{}, nil
 		}
 
+		// Check if we can accept a new task BEFORE doing expensive GetExposed operation
+		// This prevents race condition where multiple tasks waste time on GetExposed
+		// when the concurrent limit is already reached
+		if !r.dataPathMgr.CanAcceptNewTask(false) {
+			log.Debug("Data path concurrent limit reached, requeue later")
+			return ctrl.Result{Requeue: true, RequeueAfter: time.Second * 5}, nil
+		}
+
 		result, err := r.restoreExposer.GetExposed(ctx, getDataDownloadOwnerObject(dd), r.client, r.nodeName, dd.Spec.OperationTimeout.Duration)
 		if err != nil {
 			return r.errorOut(ctx, dd, err, "restore exposer is not ready", log)

pkg/controller/data_upload_controller.go

@@ -319,6 +319,15 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 			log.Info("Cancellable data path is already started")
 			return ctrl.Result{}, nil
 		}
+
+		// Check if we can accept a new task BEFORE doing expensive GetExposed operation
+		// This prevents race condition where multiple tasks waste time on GetExposed
+		// when the concurrent limit is already reached
+		if !r.dataPathMgr.CanAcceptNewTask(false) {
+			log.Debug("Data path concurrent limit reached, requeue later")
+			return ctrl.Result{Requeue: true, RequeueAfter: time.Second * 5}, nil
+		}
+
 		waitExposePara := r.setupWaitExposePara(du)
 		res, err := ep.GetExposed(ctx, getOwnerObject(du), du.Spec.OperationTimeout.Duration, waitExposePara)
 		if err != nil {

pkg/controller/pod_volume_backup_controller.go

@@ -269,6 +269,14 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 			return ctrl.Result{}, nil
 		}
 
+		// Check if we can accept a new task BEFORE doing expensive GetExposed operation
+		// This prevents race condition where multiple tasks waste time on GetExposed
+		// when the concurrent limit is already reached
+		if !r.dataPathMgr.CanAcceptNewTask(false) {
+			log.Debug("Data path concurrent limit reached, requeue later")
+			return ctrl.Result{Requeue: true, RequeueAfter: time.Second * 5}, nil
+		}
+
 		res, err := r.exposer.GetExposed(ctx, getPVBOwnerObject(pvb), r.client, r.nodeName, r.resourceTimeout)
 		if err != nil {
 			return r.errorOut(ctx, pvb, err, "exposed PVB is not ready", log)

pkg/controller/pod_volume_restore_controller.go

@@ -282,6 +282,14 @@ func (r *PodVolumeRestoreReconciler) Reconcile(ctx context.Context, req ctrl.Req
 			return ctrl.Result{}, nil
 		}
 
+		// Check if we can accept a new task BEFORE doing expensive GetExposed operation
+		// This prevents race condition where multiple tasks waste time on GetExposed
+		// when the concurrent limit is already reached
+		if !r.dataPathMgr.CanAcceptNewTask(false) {
+			log.Debug("Data path concurrent limit reached, requeue later")
+			return ctrl.Result{Requeue: true, RequeueAfter: time.Second * 5}, nil
+		}
+
 		res, err := r.exposer.GetExposed(ctx, getPVROwnerObject(pvr), r.client, r.nodeName, r.resourceTimeout)
 		if err != nil {
 			return r.errorOut(ctx, pvr, err, "exposed PVR is not ready", log)

pkg/datapath/manager.go

@@ -95,3 +95,16 @@ func (m *Manager) GetAsyncBR(jobName string) AsyncBR {
 		return nil
 	}
 }
+
+// CanAcceptNewTask checks if a new task can be accepted based on the concurrent limit.
+// This is a lightweight check that doesn't create any resources.
+func (m *Manager) CanAcceptNewTask(resume bool) bool {
+	m.trackerLock.Lock()
+	defer m.trackerLock.Unlock()
+
+	if resume {
+		return true
+	}
+
+	return len(m.tracker) < m.cocurrentNum
+}