Fix concurrent map writes crash in pvcPvMap

BackupVolumesInformation lives on the shared backupRequest and every
ItemBlock worker goroutine reaches pvcPvMap.insert via addVolumeInfo
whenever it backs up a PersistentVolume. The map had no synchronization,
so with --item-block-worker-count greater than 1 and a backup containing
multiple PVs, two workers could write the map concurrently. That is an
unrecoverable Go runtime fatal error, which kills the Velero server and
abandons the in-flight backup.

Guard pvcPvMap.data with an RWMutex so insert and retrieve are safe for
concurrent use, and guard the lazy initialization in InsertPVMap so
concurrent callers cannot each install their own map and lose entries.

The restore-side RestoreVolumeInfoTracker already locks around
pvPvc.insert. This brings the backup path in line.

Adds a regression test that reproduces the crash under -race without the
fix.

Fixes #10634

Signed-off-by: Shubham Pampattiwar <spampatt@redhat.com>

changelogs/unreleased/10635-shubham-pampattiwar

@@ -0,0 +1 @@
+Fix concurrent map writes crash in pvcPvMap when multiple ItemBlock workers back up PVs at the same time

internal/volume/volumes_information.go

@@ -411,6 +411,9 @@ type SkippedVolume struct {
 // BackupVolumesInformation contains the information needs by generating
 // the backup BackupVolumeInfo array.
 type BackupVolumesInformation struct {
+	// initLock guards the lazy initialization of pvMap, which may be triggered
+	// concurrently by multiple ItemBlock workers.
+	initLock sync.Mutex
 	// A map contains the backup-included PV detail content. The key is PV name.
 	pvMap       *pvcPvMap
 	volumeInfos []*BackupVolumeInfo
@@ -440,11 +443,17 @@ func (v *BackupVolumesInformation) Init() {
 	v.volumeInfos = make([]*BackupVolumeInfo, 0)
 }
 
+// InsertPVMap records the PV and its bound PVC in the PV map. It is called from
+// the ItemBlock worker goroutines, so it must stay safe for concurrent use.
 func (v *BackupVolumesInformation) InsertPVMap(pv corev1api.PersistentVolume, pvcName, pvcNamespace string) {
+	v.initLock.Lock()
 	if v.pvMap == nil {
 		v.Init()
 	}
-	v.pvMap.insert(pv, pvcName, pvcNamespace)
+	pvMap := v.pvMap
+	v.initLock.Unlock()
+
+	pvMap.insert(pv, pvcName, pvcNamespace)
 }
 
 func (v *BackupVolumesInformation) Result(
@@ -808,11 +817,17 @@ func (v *BackupVolumesInformation) generateVolumeInfoFromDataUpload() {
 	v.volumeInfos = append(v.volumeInfos, tmpVolumeInfos...)
 }
 
+// pvcPvMap is safe for concurrent use. On the backup path it is written by the
+// ItemBlock worker goroutines via BackupVolumesInformation.InsertPVMap.
 type pvcPvMap struct {
+	lock sync.RWMutex
 	data map[string]pvcPvInfo
 }
 
 func (m *pvcPvMap) insert(pv corev1api.PersistentVolume, pvcName, pvcNamespace string) {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+
 	m.data[pv.Name] = pvcPvInfo{
 		PVCName:      pvcName,
 		PVCNamespace: pvcNamespace,
@@ -821,6 +836,9 @@ func (m *pvcPvMap) insert(pv corev1api.PersistentVolume, pvcName, pvcNamespace s
 }
 
 func (m *pvcPvMap) retrieve(pvName, pvcName, pvcNS string) *pvcPvInfo {
+	m.lock.RLock()
+	defer m.lock.RUnlock()
+
 	if pvName != "" {
 		if info, ok := m.data[pvName]; ok {
 			return &info

internal/volume/volumes_information_test.go

@@ -18,6 +18,7 @@ package volume
 
 import (
 	"encoding/json"
+	"fmt"
 	"sync"
 	"testing"
 
@@ -1404,6 +1405,85 @@ func int64Ptr(val int) *int64 {
 	return &i
 }
 
+func TestInsertPVMapConcurrent(t *testing.T) {
+	// Velero backs up ItemBlocks on a pool of worker goroutines that all share
+	// the same BackupVolumesInformation, so InsertPVMap must be safe for
+	// concurrent use. Before the pvcPvMap lock was added this test reproduced
+	// the "fatal error: concurrent map writes" crash reported in #10634.
+	const (
+		workers    = 10
+		pvsPerWork = 50
+	)
+
+	volumesInfo := new(BackupVolumesInformation)
+	volumesInfo.Init()
+
+	wg := new(sync.WaitGroup)
+	start := make(chan struct{})
+
+	for i := range workers {
+		wg.Add(1)
+		go func(worker int) {
+			defer wg.Done()
+			<-start
+			for j := range pvsPerWork {
+				pvName := fmt.Sprintf("pv-%d-%d", worker, j)
+				volumesInfo.InsertPVMap(
+					*builder.ForPersistentVolume(pvName).ClaimRef("ns", "pvc-"+pvName).Result(),
+					"pvc-"+pvName,
+					"ns",
+				)
+				// Read concurrently as well, to exercise the RLock path.
+				volumesInfo.pvMap.retrieve(pvName, "", "")
+			}
+		}(i)
+	}
+
+	close(start)
+	wg.Wait()
+
+	require.Len(t, volumesInfo.pvMap.data, workers*pvsPerWork)
+	for i := range workers {
+		for j := range pvsPerWork {
+			pvName := fmt.Sprintf("pv-%d-%d", i, j)
+			info := volumesInfo.pvMap.retrieve(pvName, "", "")
+			require.NotNil(t, info, "PV %s missing from the map", pvName)
+			require.Equal(t, "pvc-"+pvName, info.PVCName)
+			require.Equal(t, "ns", info.PVCNamespace)
+		}
+	}
+}
+
+func TestInsertPVMapLazyInitConcurrent(t *testing.T) {
+	// InsertPVMap lazily initializes pvMap when Init has not been called.
+	// Concurrent callers must not each install their own map and lose entries.
+	const workers = 20
+
+	volumesInfo := new(BackupVolumesInformation)
+
+	wg := new(sync.WaitGroup)
+	start := make(chan struct{})
+
+	for i := range workers {
+		wg.Add(1)
+		go func(worker int) {
+			defer wg.Done()
+			<-start
+			pvName := fmt.Sprintf("pv-%d", worker)
+			volumesInfo.InsertPVMap(
+				*builder.ForPersistentVolume(pvName).ClaimRef("ns", "pvc-"+pvName).Result(),
+				"pvc-"+pvName,
+				"ns",
+			)
+		}(i)
+	}
+
+	close(start)
+	wg.Wait()
+
+	require.Len(t, volumesInfo.pvMap.data, workers)
+}
+
 func TestGetVolumeSnapshotClasses(t *testing.T) {
 	class := &snapshotv1api.VolumeSnapshotClass{
 		ObjectMeta: metav1.ObjectMeta{