[AI-generated]Fix race condition: check concurrent limit before GetExposed

[kvaps]

Fix race condition in data upload controller concurrent limit check

Summary

Fixed a race condition where multiple DataUpload tasks were wasting time on expensive GetExposed operations even when the concurrent limit was already reached. This caused all tasks to get ConcurrentLimitExceed errors after spending time on GetExposed, leading to inefficient processing and tasks stuck in Prepared state.

Changes

1. Added CanAcceptNewTask() method to datapath.Manager (pkg/datapath/manager.go):

   • Lightweight check that verifies if a new task can be accepted based on concurrent limit
   • Thread-safe with mutex protection
   • Returns true if len(tracker) < cocurrentNum

2. Added concurrent limit check BEFORE GetExposed in all controllers:

   • DataUploadReconciler (pkg/controller/data_upload_controller.go)
   • DataDownloadReconciler (pkg/controller/data_download_controller.go)
   • PodVolumeBackupReconciler (pkg/controller/pod_volume_backup_controller.go)
   • PodVolumeRestoreReconciler (pkg/controller/pod_volume_restore_controller.go)

Problem

Previously, the concurrent limit was checked AFTER executing the expensive GetExposed operation:

1. Multiple tasks would check GetAsyncBR(taskName) → all get nil (their names not in tracker yet)
2. All tasks would execute GetExposed (expensive, can take seconds)
3. All tasks would reach CreateMicroServiceBRWatcher simultaneously
4. Only the first task would succeed (limit = 1), others get ConcurrentLimitExceed
5. Error was logged at DEBUG level, so not visible in normal logs
6. Tasks would requeue and repeat the cycle

This meant tasks were wasting CPU and time on GetExposed even when the concurrent limit was already reached, causing inefficiency and delays.

Solution

The fix adds an early check for the concurrent limit BEFORE executing GetExposed:

// Check if we can accept a new task BEFORE doing expensive GetExposed operation
if !r.dataPathMgr.CanAcceptNewTask(false) {
    log.Debug("Data path concurrent limit reached, requeue later")
    return ctrl.Result{Requeue: true, RequeueAfter: time.Second * 5}, nil
}

Now tasks immediately requeue if the limit is reached, without wasting resources on GetExposed.

Benefits

• Resource efficiency: Tasks don't waste time on GetExposed when limit is reached
• Fairer queue: Tasks check limit earlier and requeue, freeing resources faster
• Fewer race conditions: Limit check happens before expensive operations
• Universal fix: Applied to all controllers sharing the common tracker

Testing

The change is thread-safe as CanAcceptNewTask uses the same mutex as other tracker operations. The logic matches the existing check in CreateMicroServiceBRWatcher but allows early exit before expensive operations.

Does your change fix a particular issue?

Yes, this fixes a race condition that causes DataUpload tasks to waste resources and get stuck when concurrent limit is reached.

Please indicate you've done the following:

• Accepted the DCO. Commits without the DCO will delay acceptance.
• Created a changelog file (make new-changelog) or comment /kind changelog-not-required on this PR.
• Updated the corresponding documentation in site/content/docs/main.

[Lyndon-Li]

Have you tested it with large number of tasks and prove the effect? I think current changes won't work well for the expectation.

As this flow mentioned above:

1. Multiple tasks would check GetAsyncBR(taskName) → all get nil (their names not in tracker yet)
2. All tasks would execute GetExposed (expensive, can take seconds)

Even we have the current change:

1. Multiple tasks would check GetAsyncBR(taskName) → all get nil (their names not in tracker yet)
2. All tasks call r.dataPathMgr.CanAcceptNewTask. Since GetExposed is time consuming and len(m.tracker) is increased ONLY AFTER GetExposed in CreateMicroServiceBRWatcher, most probably, all tasks get passed by CanAcceptNewTask
3. Then STILL all tasks call GetExposed

[kvaps]

@Lyndon-Li sorry you talk with my AI agent 😊 I provide reponse bellow:

---

You're absolutely right! My initial fix didn't fully solve the race condition. The issue is that CanAcceptNewTask() and CreateMicroServiceBRWatcher() are separate non-atomic operations, and multiple tasks can pass the check simultaneously before any of them completes GetExposed and adds to the tracker.

I've now implemented a reservation system that atomically reserves slots in the tracker before GetExposed:

Changes

1. Added ReserveSlot(taskName) method - Atomically reserves a slot by adding to a reservations map, which is counted along with the tracker for concurrency limits
2. Added ReleaseReservation(taskName) method - Releases a reserved slot if GetExposed or subsequent operations fail
3. Modified CreateMicroServiceBRWatcher - Automatically releases the reservation when adding the actual AsyncBR to tracker
4. Updated all controllers - Now call ReserveSlot() BEFORE GetExposed, and ReleaseReservation() on any error path

How it works

1. Task calls ReserveSlot(taskName) → atomically checks len(tracker) + len(reservations) < cocurrentNum and adds to reservations map
2. Task proceeds with GetExposed() - only if reservation succeeded
3. If GetExposed succeeds → CreateMicroServiceBRWatcher moves from reservations to tracker atomically
4. If GetExposed fails → ReleaseReservation removes from reservations

This ensures that:

• Only tasks with successfully reserved slots proceed to GetExposed
• The concurrent limit check is atomic (protected by mutex)
• Other tasks see len(tracker) + len(reservations) >= cocurrentNum immediately and requeue
• No wasted CPU/time on GetExposed when limit is reached

The reservation is counted in the concurrency limit, so if cocurrentNum=1, only one task can reserve a slot at a time, preventing the race condition completely.

Context

In my cluster, I have 465 DataUpload tasks total, with:

• 212 Completed
• 252 Canceled (many due to the race condition issue)
• 1 Failed

The race condition was particularly problematic with cocurrentNum=1 and large number of tasks, as all tasks would repeatedly try to execute GetExposed simultaneously, leading to wasted resources and tasks getting stuck in Prepared state before eventually being canceled.

[Lyndon-Li]

I just thought about it further, there is no way that the problem you mentioned could happen, because there is only one reconciler thread, or in another word, the tasks are actually processed synchronously in each node.
The asynchronization comes from the asynchronous processing of InProgress tasks.

[kvaps]

@Lyndon-Li I just had the issue when all my datauploads were in Prepared state and no active InProggress jobs.
Velero-agents were succefully running for 30 days, and do nothing with them.

After I restarted, they started accpting new jobs and some of Prepared jobs become InProgress, later whole queue was successfully finished

[Lyndon-Li]

@Lyndon-Li I just had the issue when all my datauploads were in Prepared state and no active InProggress jobs. Velero-agents were succefully running for 30 days, and do nothing with them.

After I restarted, they started accpting new jobs and some of Prepared jobs become InProgress, later whole queue was successfully finished

This stuck problem looks different from the problem the PR is try to solve --- CPU time waste.
For the stuck problem, please open an issue, see #9449 (comment).

For CPU time waste:

1. We cannot periodically enqueue Prepared tasks, this makes the situation even worse
2. GetExposed is not very time consuming, when the task is ready, GetExposed means a few API server calls only
3. Tasks in Prepared phases won't squeeze one another as they are processed synchronously

[kvaps]

Sure, done #9453