From da5bee7097fb38cfbe4e75d94f3c8e66d8be5139 Mon Sep 17 00:00:00 2001 From: lyndon-li <98304688+Lyndon-Li@users.noreply.github.com> Date: Mon, 17 Aug 2026 16:55:01 +0800 Subject: [PATCH] Use thread safe map for cancel recorder (#10255) * use thread safe map for cancel recorder Signed-off-by: Lyndon-Li * use atomic load and store Signed-off-by: Lyndon-Li --------- Signed-off-by: Lyndon-Li --- changelogs/unreleased/10255-Lyndon-Li | 1 + pkg/controller/data_download_controller.go | 16 ++--- .../data_download_controller_test.go | 62 ++++++++++++++++++- pkg/controller/data_upload_controller.go | 16 ++--- pkg/controller/data_upload_controller_test.go | 60 +++++++++++++++++- .../pod_volume_backup_controller.go | 16 ++--- .../pod_volume_backup_controller_test.go | 62 ++++++++++++++++++- .../pod_volume_restore_controller.go | 16 ++--- .../pod_volume_restore_controller_test.go | 59 +++++++++++++++++- 9 files changed, 264 insertions(+), 44 deletions(-) create mode 100644 changelogs/unreleased/10255-Lyndon-Li diff --git a/changelogs/unreleased/10255-Lyndon-Li b/changelogs/unreleased/10255-Lyndon-Li new file mode 100644 index 000000000..f6ddc1e76 --- /dev/null +++ b/changelogs/unreleased/10255-Lyndon-Li @@ -0,0 +1 @@ +Use thread safe map for cancel recorder \ No newline at end of file diff --git a/pkg/controller/data_download_controller.go b/pkg/controller/data_download_controller.go index 422879d6e..19d788f3f 100644 --- a/pkg/controller/data_download_controller.go +++ b/pkg/controller/data_download_controller.go @@ -20,6 +20,7 @@ import ( "context" "fmt" "strings" + "sync" "time" "github.com/cockroachdb/errors" @@ -74,7 +75,7 @@ type DataDownloadReconciler struct { podResources corev1api.ResourceRequirements preparingTimeout time.Duration metrics *metrics.ServerMetrics - cancelledDataDownload map[string]time.Time + cancelledDataDownload sync.Map dataMovePriorityClass string repoConfigMgr repository.ConfigManager podLabels map[string]string @@ -118,7 +119,6 @@ func NewDataDownloadReconciler( podResources: podResources, preparingTimeout: preparingTimeout, metrics: metrics, - cancelledDataDownload: make(map[string]time.Time), dataMovePriorityClass: dataMovePriorityClass, repoConfigMgr: repoConfigMgr, podLabels: podLabels, @@ -198,7 +198,7 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request } } } else { - delete(r.cancelledDataDownload, dd.Name) + r.cancelledDataDownload.Delete(dd.Name) // put the finalizer remove action here for all cr will goes to the final status, we could check finalizer and do remove action in final status // instead of intermediate state. @@ -223,9 +223,9 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request } if dd.Spec.Cancel { - if spotted, found := r.cancelledDataDownload[dd.Name]; !found { - r.cancelledDataDownload[dd.Name] = r.Clock.Now() - } else { + v, loaded := r.cancelledDataDownload.LoadOrStore(dd.Name, r.Clock.Now()) + if loaded { + spotted := v.(time.Time) delay := cancelDelayOthers if dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseInProgress { delay = cancelDelayInProgress @@ -234,7 +234,7 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request if time.Since(spotted) > delay { log.Infof("Data download %s is canceled in Phase %s but not handled in rasonable time", dd.GetName(), dd.Status.Phase) if r.tryCancelDataDownload(ctx, dd, "") { - delete(r.cancelledDataDownload, dd.Name) + r.cancelledDataDownload.Delete(dd.Name) } return ctrl.Result{}, nil @@ -556,7 +556,7 @@ func (r *DataDownloadReconciler) OnDataDownloadCancelled(ctx context.Context, na log.WithError(err).Error("error updating data download status") } else { r.metrics.RegisterDataDownloadCancel(r.nodeName) - delete(r.cancelledDataDownload, dd.Name) + r.cancelledDataDownload.Delete(dd.Name) } } diff --git a/pkg/controller/data_download_controller_test.go b/pkg/controller/data_download_controller_test.go index a605fcaaa..4ef79b823 100644 --- a/pkg/controller/data_download_controller_test.go +++ b/pkg/controller/data_download_controller_test.go @@ -19,9 +19,12 @@ package controller import ( "context" "fmt" + "sync" "testing" "time" + clocktesting "k8s.io/utils/clock/testing" + "github.com/cockroachdb/errors" "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" @@ -507,7 +510,7 @@ func TestDataDownloadReconcile(t *testing.T) { } if test.sportTime != nil { - r.cancelledDataDownload[test.dd.Name] = test.sportTime.Time + r.cancelledDataDownload.Store(test.dd.Name, test.sportTime.Time) } if test.constrained { @@ -624,9 +627,15 @@ func TestDataDownloadReconcile(t *testing.T) { } if test.expectCancelRecord { - assert.Contains(t, r.cancelledDataDownload, test.dd.Name) + _, ok := r.cancelledDataDownload.Load(test.dd.Name) + assert.True(t, ok) } else { - assert.Empty(t, r.cancelledDataDownload) + empty := true + r.cancelledDataDownload.Range(func(key, value any) bool { + empty = false + return false + }) + assert.True(t, empty) } if isDataDownloadInFinalState(&dd) || dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseInProgress { @@ -1437,3 +1446,50 @@ func TestDataDownloadSetupExposeParam(t *testing.T) { }) } } + +type sequenceClock struct { + *clocktesting.FakeClock + mu sync.Mutex +} + +func (c *sequenceClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + c.FakeClock.Step(time.Second) + return c.FakeClock.Now() +} + +func TestDataDownloadCancelConcurrency(t *testing.T) { + ctx := t.Context() + dd := dataDownloadBuilder().Cancel(true).Phase(velerov2alpha1api.DataDownloadPhaseInProgress).Result() + + r, err := initDataDownloadReconciler(t, nil) + require.NoError(t, err) + + err = r.client.Create(ctx, dd) + require.NoError(t, err) + + firstTime := time.Now() + // manually store the initial time + r.cancelledDataDownload.Store(dd.Name, firstTime) + + // Custom clock that returns a different time each call + r.Clock = &sequenceClock{FakeClock: clocktesting.NewFakeClock(firstTime)} + + var wg sync.WaitGroup + routines := 50 + wg.Add(routines) + + for i := 0; i < routines; i++ { + go func() { + defer wg.Done() + _, _ = r.Reconcile(ctx, ctrl.Request{NamespacedName: types.NamespacedName{Name: dd.Name, Namespace: dd.Namespace}}) + }() + } + + wg.Wait() + + v, ok := r.cancelledDataDownload.Load(dd.Name) + assert.True(t, ok) + assert.Equal(t, firstTime, v.(time.Time), "The initially recorded timestamp should be preserved") +} diff --git a/pkg/controller/data_upload_controller.go b/pkg/controller/data_upload_controller.go index 61ccfef58..e7eaff956 100644 --- a/pkg/controller/data_upload_controller.go +++ b/pkg/controller/data_upload_controller.go @@ -20,6 +20,7 @@ import ( "context" "fmt" "strings" + "sync" "time" "github.com/cockroachdb/errors" @@ -81,7 +82,7 @@ type DataUploadReconciler struct { podResources corev1api.ResourceRequirements preparingTimeout time.Duration metrics *metrics.ServerMetrics - cancelledDataUpload map[string]time.Time + cancelledDataUpload sync.Map dataMovePriorityClass string podLabels map[string]string podAnnotations map[string]string @@ -130,7 +131,6 @@ func NewDataUploadReconciler( podResources: podResources, preparingTimeout: preparingTimeout, metrics: metrics, - cancelledDataUpload: make(map[string]time.Time), dataMovePriorityClass: dataMovePriorityClass, podLabels: podLabels, podAnnotations: podAnnotations, @@ -207,7 +207,7 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request) } } } else { - delete(r.cancelledDataUpload, du.Name) + r.cancelledDataUpload.Delete(du.Name) // put the finalizer remove action here for all cr will goes to the final status, we could check finalizer and do remove action in final status // instead of intermediate state. @@ -232,9 +232,9 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request) } if du.Spec.Cancel { - if spotted, found := r.cancelledDataUpload[du.Name]; !found { - r.cancelledDataUpload[du.Name] = r.Clock.Now() - } else { + v, loaded := r.cancelledDataUpload.LoadOrStore(du.Name, r.Clock.Now()) + if loaded { + spotted := v.(time.Time) delay := cancelDelayOthers if du.Status.Phase == velerov2alpha1api.DataUploadPhaseInProgress { delay = cancelDelayInProgress @@ -243,7 +243,7 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request) if time.Since(spotted) > delay { log.Infof("Data upload %s is canceled in Phase %s but not handled in reasonable time", du.GetName(), du.Status.Phase) if r.tryCancelDataUpload(ctx, du, "") { - delete(r.cancelledDataUpload, du.Name) + r.cancelledDataUpload.Delete(du.Name) } return ctrl.Result{}, nil @@ -577,7 +577,7 @@ func (r *DataUploadReconciler) OnDataUploadCancelled(ctx context.Context, namesp log.WithError(err).Error("error updating DataUpload status") } else { r.metrics.RegisterDataUploadCancel(r.nodeName) - delete(r.cancelledDataUpload, du.Name) + r.cancelledDataUpload.Delete(du.Name) } } diff --git a/pkg/controller/data_upload_controller_test.go b/pkg/controller/data_upload_controller_test.go index ec819f8eb..30b5926ac 100644 --- a/pkg/controller/data_upload_controller_test.go +++ b/pkg/controller/data_upload_controller_test.go @@ -19,6 +19,7 @@ package controller import ( "context" "fmt" + "sync" "testing" "time" @@ -672,7 +673,7 @@ func TestReconcile(t *testing.T) { } if test.sportTime != nil { - r.cancelledDataUpload[test.du.Name] = test.sportTime.Time + r.cancelledDataUpload.Store(test.du.Name, test.sportTime.Time) } if test.constrained { @@ -752,9 +753,15 @@ func TestReconcile(t *testing.T) { } if test.expectCancelRecord { - assert.Contains(t, r.cancelledDataUpload, test.du.Name) + _, ok := r.cancelledDataUpload.Load(test.du.Name) + assert.True(t, ok) } else { - assert.Empty(t, r.cancelledDataUpload) + empty := true + r.cancelledDataUpload.Range(func(key, value any) bool { + empty = false + return false + }) + assert.True(t, empty) } if isDataUploadInFinalState(&du) || du.Status.Phase == velerov2alpha1api.DataUploadPhaseInProgress { @@ -1561,3 +1568,50 @@ func TestDataUploadSetupExposeParam(t *testing.T) { }) } } + +type dataUploadSequenceClock struct { + *testclocks.FakeClock + mu sync.Mutex +} + +func (c *dataUploadSequenceClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + c.FakeClock.Step(time.Second) + return c.FakeClock.Now() +} + +func TestDataUploadCancelConcurrency(t *testing.T) { + ctx := t.Context() + du := dataUploadBuilder().Cancel(true).Phase(velerov2alpha1api.DataUploadPhaseInProgress).Result() + + r, err := initDataUploaderReconciler() + require.NoError(t, err) + + err = r.client.Create(ctx, du) + require.NoError(t, err) + + firstTime := time.Now() + // manually store the initial time + r.cancelledDataUpload.Store(du.Name, firstTime) + + // Custom clock that returns a different time each call + r.Clock = &dataUploadSequenceClock{FakeClock: testclocks.NewFakeClock(firstTime)} + + var wg sync.WaitGroup + routines := 50 + wg.Add(routines) + + for i := 0; i < routines; i++ { + go func() { + defer wg.Done() + _, _ = r.Reconcile(ctx, ctrl.Request{NamespacedName: types.NamespacedName{Name: du.Name, Namespace: du.Namespace}}) + }() + } + + wg.Wait() + + v, ok := r.cancelledDataUpload.Load(du.Name) + assert.True(t, ok) + assert.Equal(t, firstTime, v.(time.Time), "The initially recorded timestamp should be preserved") +} diff --git a/pkg/controller/pod_volume_backup_controller.go b/pkg/controller/pod_volume_backup_controller.go index 2372bf25b..13dbd5d79 100644 --- a/pkg/controller/pod_volume_backup_controller.go +++ b/pkg/controller/pod_volume_backup_controller.go @@ -20,6 +20,7 @@ import ( "context" "fmt" "strings" + "sync" "time" "github.com/cockroachdb/errors" @@ -89,7 +90,6 @@ func NewPodVolumeBackupReconciler( preparingTimeout: preparingTimeout, resourceTimeout: resourceTimeout, exposer: exposer.NewPodVolumeExposer(kubeClient, logger), - cancelledPVB: make(map[string]time.Time), dataMovePriorityClass: dataMovePriorityClass, privileged: privileged, podLabels: podLabels, @@ -112,7 +112,7 @@ type PodVolumeBackupReconciler struct { vgdpCounter *exposer.VgdpCounter preparingTimeout time.Duration resourceTimeout time.Duration - cancelledPVB map[string]time.Time + cancelledPVB sync.Map dataMovePriorityClass string privileged bool podLabels map[string]string @@ -183,7 +183,7 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ } } } else { - delete(r.cancelledPVB, pvb.Name) + r.cancelledPVB.Delete(pvb.Name) if controllerutil.ContainsFinalizer(pvb, PodVolumeFinalizer) { if err := UpdatePVBWithRetry(ctx, r.client, req.NamespacedName, log, func(pvb *velerov1api.PodVolumeBackup) bool { @@ -204,9 +204,9 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ } if pvb.Spec.Cancel { - if spotted, found := r.cancelledPVB[pvb.Name]; !found { - r.cancelledPVB[pvb.Name] = r.clock.Now() - } else { + v, loaded := r.cancelledPVB.LoadOrStore(pvb.Name, r.clock.Now()) + if loaded { + spotted := v.(time.Time) delay := cancelDelayOthers if pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseInProgress { delay = cancelDelayInProgress @@ -215,7 +215,7 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ if time.Since(spotted) > delay { log.Infof("PVB %s is canceled in Phase %s but not handled in reasonable time", pvb.GetName(), pvb.Status.Phase) if r.tryCancelPodVolumeBackup(ctx, pvb, "") { - delete(r.cancelledPVB, pvb.Name) + r.cancelledPVB.Delete(pvb.Name) } return ctrl.Result{}, nil @@ -620,7 +620,7 @@ func (r *PodVolumeBackupReconciler) OnDataPathCancelled(ctx context.Context, nam }); err != nil { log.WithError(err).Error("error updating PVB status on cancel") } else { - delete(r.cancelledPVB, pvb.Name) + r.cancelledPVB.Delete(pvb.Name) } } diff --git a/pkg/controller/pod_volume_backup_controller_test.go b/pkg/controller/pod_volume_backup_controller_test.go index 8b05f0e3b..21e30d5db 100644 --- a/pkg/controller/pod_volume_backup_controller_test.go +++ b/pkg/controller/pod_volume_backup_controller_test.go @@ -19,9 +19,12 @@ package controller import ( "context" "fmt" + "sync" "testing" "time" + clocktesting "k8s.io/utils/clock/testing" + "github.com/cockroachdb/errors" "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" @@ -489,7 +492,7 @@ func TestPVBReconcile(t *testing.T) { } if test.sportTime != nil { - r.cancelledPVB[test.pvb.Name] = test.sportTime.Time + r.cancelledPVB.Store(test.pvb.Name, test.sportTime.Time) } if test.constrained { @@ -567,9 +570,15 @@ func TestPVBReconcile(t *testing.T) { } if test.expectCancelRecord { - assert.Contains(t, r.cancelledPVB, test.pvb.Name) + _, ok := r.cancelledPVB.Load(test.pvb.Name) + assert.True(t, ok) } else { - assert.Empty(t, r.cancelledPVB) + empty := true + r.cancelledPVB.Range(func(key, value any) bool { + empty = false + return false + }) + assert.True(t, empty) } if isPVBInFinalState(&pvb) || pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseInProgress { @@ -1308,3 +1317,50 @@ func TestPodVolumeBackupSetupExposeParam(t *testing.T) { }) } } + +type pvbSequenceClock struct { + *clocktesting.FakeClock + mu sync.Mutex +} + +func (c *pvbSequenceClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + c.FakeClock.Step(time.Second) + return c.FakeClock.Now() +} + +func TestPodVolumeBackupCancelConcurrency(t *testing.T) { + ctx := t.Context() + pvb := builder.ForPodVolumeBackup(velerov1api.DefaultNamespace, "pvb-1").Cancel(true).Phase(velerov1api.PodVolumeBackupPhaseInProgress).Result() + + r, err := initPVBReconciler() + require.NoError(t, err) + + err = r.client.Create(ctx, pvb) + require.NoError(t, err) + + firstTime := time.Now() + // manually store the initial time + r.cancelledPVB.Store(pvb.Name, firstTime) + + // Custom clock that returns a different time each call + r.clock = &pvbSequenceClock{FakeClock: clocktesting.NewFakeClock(firstTime)} + + var wg sync.WaitGroup + routines := 50 + wg.Add(routines) + + for i := 0; i < routines; i++ { + go func() { + defer wg.Done() + _, _ = r.Reconcile(ctx, ctrl.Request{NamespacedName: types.NamespacedName{Name: pvb.Name, Namespace: pvb.Namespace}}) + }() + } + + wg.Wait() + + v, ok := r.cancelledPVB.Load(pvb.Name) + assert.True(t, ok) + assert.Equal(t, firstTime, v.(time.Time), "The initially recorded timestamp should be preserved") +} diff --git a/pkg/controller/pod_volume_restore_controller.go b/pkg/controller/pod_volume_restore_controller.go index 159598dca..b6d4985fa 100644 --- a/pkg/controller/pod_volume_restore_controller.go +++ b/pkg/controller/pod_volume_restore_controller.go @@ -20,6 +20,7 @@ import ( "context" "fmt" "strings" + "sync" "time" "github.com/cockroachdb/errors" @@ -90,7 +91,6 @@ func NewPodVolumeRestoreReconciler( preparingTimeout: preparingTimeout, resourceTimeout: resourceTimeout, exposer: exposer.NewPodVolumeExposer(kubeClient, logger), - cancelledPVR: make(map[string]time.Time), dataMovePriorityClass: dataMovePriorityClass, privileged: privileged, repoConfigMgr: repoConfigMgr, @@ -114,7 +114,7 @@ type PodVolumeRestoreReconciler struct { vgdpCounter *exposer.VgdpCounter preparingTimeout time.Duration resourceTimeout time.Duration - cancelledPVR map[string]time.Time + cancelledPVR sync.Map dataMovePriorityClass string privileged bool repoConfigMgr repository.ConfigManager @@ -188,7 +188,7 @@ func (r *PodVolumeRestoreReconciler) Reconcile(ctx context.Context, req ctrl.Req } } } else { - delete(r.cancelledPVR, pvr.Name) + r.cancelledPVR.Delete(pvr.Name) if controllerutil.ContainsFinalizer(pvr, PodVolumeFinalizer) { if err := UpdatePVRWithRetry(ctx, r.client, req.NamespacedName, log, func(pvr *velerov1api.PodVolumeRestore) bool { @@ -209,9 +209,9 @@ func (r *PodVolumeRestoreReconciler) Reconcile(ctx context.Context, req ctrl.Req } if pvr.Spec.Cancel { - if spotted, found := r.cancelledPVR[pvr.Name]; !found { - r.cancelledPVR[pvr.Name] = r.clock.Now() - } else { + v, loaded := r.cancelledPVR.LoadOrStore(pvr.Name, r.clock.Now()) + if loaded { + spotted := v.(time.Time) delay := cancelDelayOthers if pvr.Status.Phase == velerov1api.PodVolumeRestorePhaseInProgress { delay = cancelDelayInProgress @@ -220,7 +220,7 @@ func (r *PodVolumeRestoreReconciler) Reconcile(ctx context.Context, req ctrl.Req if time.Since(spotted) > delay { log.Infof("PVR %s is canceled in Phase %s but not handled in rasonable time", pvr.GetName(), pvr.Status.Phase) if r.tryCancelPodVolumeRestore(ctx, pvr, "") { - delete(r.cancelledPVR, pvr.Name) + r.cancelledPVR.Delete(pvr.Name) } return ctrl.Result{}, nil @@ -895,7 +895,7 @@ func (r *PodVolumeRestoreReconciler) OnDataPathCancelled(ctx context.Context, na }); err != nil { log.WithError(err).Error("error updating PVR status on cancel") } else { - delete(r.cancelledPVR, pvr.Name) + r.cancelledPVR.Delete(pvr.Name) } } diff --git a/pkg/controller/pod_volume_restore_controller_test.go b/pkg/controller/pod_volume_restore_controller_test.go index 8ec1f7eca..73167c76f 100644 --- a/pkg/controller/pod_volume_restore_controller_test.go +++ b/pkg/controller/pod_volume_restore_controller_test.go @@ -19,9 +19,12 @@ package controller import ( "context" "fmt" + "sync" "testing" "time" + clocktesting "k8s.io/utils/clock/testing" + "github.com/cockroachdb/errors" "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" @@ -1087,7 +1090,7 @@ func TestPodVolumeRestoreReconcile(t *testing.T) { } if test.sportTime != nil { - r.cancelledPVR[test.pvr.Name] = test.sportTime.Time + r.cancelledPVR.Store(test.pvr.Name, test.sportTime.Time) } if test.constrained { @@ -1208,9 +1211,15 @@ func TestPodVolumeRestoreReconcile(t *testing.T) { } if test.expectCancelRecord { - assert.Contains(t, r.cancelledPVR, test.pvr.Name) + _, ok := r.cancelledPVR.Load(test.pvr.Name) + assert.True(t, ok) } else { - assert.Empty(t, r.cancelledPVR) + empty := true + r.cancelledPVR.Range(func(key, value any) bool { + empty = false + return false + }) + assert.True(t, empty) } if isPVRInFinalState(&pvr) || pvr.Status.Phase == velerov1api.PodVolumeRestorePhaseInProgress { @@ -1935,3 +1944,47 @@ func TestResumeCancellablePodVolumeRestore(t *testing.T) { }) } } + +type pvrSequenceClock struct { + *clocktesting.FakeClock + mu sync.Mutex +} + +func (c *pvrSequenceClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + c.FakeClock.Step(time.Second) + return c.FakeClock.Now() +} + +func TestPodVolumeRestoreCancelConcurrency(t *testing.T) { + ctx := t.Context() + pvr := builder.ForPodVolumeRestore(velerov1api.DefaultNamespace, "pvr-1").Cancel(true).Phase(velerov1api.PodVolumeRestorePhaseInProgress).Result() + + r, err := initPodVolumeRestoreReconciler(nil, []client.Object{pvr}) + require.NoError(t, err) + + firstTime := time.Now() + // manually store the initial time + r.cancelledPVR.Store(pvr.Name, firstTime) + + // Custom clock that returns a different time each call + r.clock = &pvrSequenceClock{FakeClock: clocktesting.NewFakeClock(firstTime)} + + var wg sync.WaitGroup + routines := 50 + wg.Add(routines) + + for i := 0; i < routines; i++ { + go func() { + defer wg.Done() + _, _ = r.Reconcile(ctx, ctrl.Request{NamespacedName: types.NamespacedName{Name: pvr.Name, Namespace: pvr.Namespace}}) + }() + } + + wg.Wait() + + v, ok := r.cancelledPVR.Load(pvr.Name) + assert.True(t, ok) + assert.Equal(t, firstTime, v.(time.Time), "The initially recorded timestamp should be preserved") +}