🐛 Use sync map to prevent concurrent writes. (#6053)

Signed-off-by: Preslav <preslav@mondoo.com>

Author: preslavgerchev

providers-sdk/v1/recording/asset_recording_test.go

@@ -8,12 +8,13 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"go.mondoo.com/cnquery/v12/providers-sdk/v1/inventory"
+	"go.mondoo.com/cnquery/v12/utils/syncx"
 )
 
 func TestAssetRecording(t *testing.T) {
 	t.Run("add asset by id only", func(t *testing.T) {
 		rec := &recording{
-			assets: map[uint32]*Asset{},
+			assets: syncx.Map[*Asset]{},
 			Assets: []*Asset{},
 		}
 
@@ -27,13 +28,12 @@ func TestAssetRecording(t *testing.T) {
 		}
 		rec.EnsureAsset(asset, "provider", 1, conf)
 
-		require.Len(t, rec.assets, 0)
 		require.Len(t, rec.Assets, 0)
 	})
 
 	t.Run("add asset by mrn", func(t *testing.T) {
 		rec := &recording{
-			assets: map[uint32]*Asset{},
+			assets: syncx.Map[*Asset]{},
 			Assets: []*Asset{},
 		}
 
@@ -47,7 +47,6 @@ func TestAssetRecording(t *testing.T) {
 		}
 		rec.EnsureAsset(asset, "provider", 1, conf)
 
-		require.Len(t, rec.assets, 1)
 		require.Len(t, rec.Assets, 1)
 		require.Len(t, rec.Assets[0].connections, 1)
 		require.Len(t, rec.Assets[0].Resources, 0)
@@ -59,7 +58,6 @@ func TestAssetRecording(t *testing.T) {
 		asset.Mrn = "asset-mrn"
 		asset.PlatformIds = []string{"platform-id", "asset-mrn"}
 		rec.EnsureAsset(asset, "provider", 1, conf)
-		require.Len(t, rec.assets, 1)
 		require.Len(t, rec.Assets, 1)
 		require.Len(t, rec.Assets[0].connections, 1)
 		require.Len(t, rec.Assets[0].Resources, 0)
@@ -71,7 +69,7 @@ func TestAssetRecording(t *testing.T) {
 
 	t.Run("add asset by platform id and mrn", func(t *testing.T) {
 		rec := &recording{
-			assets: map[uint32]*Asset{},
+			assets: syncx.Map[*Asset]{},
 			Assets: []*Asset{},
 		}
 
@@ -85,7 +83,6 @@ func TestAssetRecording(t *testing.T) {
 		}
 		rec.EnsureAsset(asset, "provider", 1, conf)
 
-		require.Len(t, rec.assets, 1)
 		require.Len(t, rec.Assets, 1)
 		require.Len(t, rec.Assets[0].connections, 1)
 		require.Len(t, rec.Assets[0].Resources, 0)
@@ -96,7 +93,6 @@ func TestAssetRecording(t *testing.T) {
 		// re-add again by platform id, ensure nothing gets duplicated
 		asset.Mrn = ""
 		rec.EnsureAsset(asset, "provider", 1, conf)
-		require.Len(t, rec.assets, 1)
 		require.Len(t, rec.Assets, 1)
 		require.Len(t, rec.Assets[0].connections, 1)
 		require.Len(t, rec.Assets[0].Resources, 0)
@@ -108,7 +104,6 @@ func TestAssetRecording(t *testing.T) {
 		// re-add again by mrn, ensure nothing gets duplicated
 		asset.Mrn = "asset-mrn"
 		rec.EnsureAsset(asset, "provider", 1, conf)
-		require.Len(t, rec.assets, 1)
 		require.Len(t, rec.Assets, 1)
 		require.Len(t, rec.Assets[0].connections, 1)
 		require.Len(t, rec.Assets[0].Resources, 0)

providers-sdk/v1/recording/recording.go

@@ -15,13 +15,14 @@ import (
 	"go.mondoo.com/cnquery/v12/providers-sdk/v1/inventory"
 	"go.mondoo.com/cnquery/v12/types"
 	"go.mondoo.com/cnquery/v12/utils/multierr"
+	"go.mondoo.com/cnquery/v12/utils/syncx"
 )
 
 type recording struct {
 	Assets []*Asset `json:"assets"`
 	Path   string   `json:"-"`
 	// assets is used for fast connection to asset lookup
-	assets          map[uint32]*Asset `json:"-"`
+	assets          syncx.Map[*Asset] `json:"-"`
 	prettyPrintJSON bool              `json:"-"`
 	// this mode is used when we use the recording layer for data,
 	// but not for storing it on disk
@@ -94,7 +95,7 @@ func (n *readOnly) EnsureAsset(asset *inventory.Asset, provider string, connecti
 	// we are severely lacking connection IDs.
 	existing := n.getExistingAsset(asset)
 	if existing != nil {
-		n.assets[connectionID] = existing
+		n.assets.Set(fmt.Sprintf("%d", connectionID), existing)
 	}
 }
 
@@ -142,7 +143,7 @@ func NewWithFile(path string, opts RecordingOptions) (llx.Recording, error) {
 				Path:            path,
 				prettyPrintJSON: opts.PrettyPrintJSON,
 				doNotSave:       opts.DoNotSave,
-				assets:          map[uint32]*Asset{},
+				assets:          syncx.Map[*Asset]{},
 			}
 			res.refreshCache() // only for initialization
 			return res, nil
@@ -203,7 +204,7 @@ func (r *recording) Save() error {
 }
 
 func (r *recording) refreshCache() {
-	r.assets = make(map[uint32]*Asset, len(r.Assets))
+	r.assets = syncx.Map[*Asset]{}
 	for i := range r.Assets {
 		asset := r.Assets[i]
 		asset.RefreshCache()
@@ -213,7 +214,7 @@ func (r *recording) refreshCache() {
 			// initially load this object, so we won't know yet which asset belongs
 			// to which connection.
 			if conn.Id != 0 {
-				r.assets[conn.Id] = asset
+				r.assets.Set(fmt.Sprintf("%d", conn.Id), asset)
 			}
 		}
 	}
@@ -379,11 +380,11 @@ func (r *recording) EnsureAsset(asset *inventory.Asset, providerID string, conne
 		Connector:  conf.Type,
 		Id:         conf.Id,
 	}
-	r.assets[connectionID] = recordingAsset
+	r.assets.Set(fmt.Sprintf("%d", conf.Id), recordingAsset)
 }
 
 func (r *recording) AddData(connectionID uint32, resource string, id string, field string, data *llx.RawData) {
-	asset, ok := r.assets[connectionID]
+	asset, ok := r.assets.Get(fmt.Sprintf("%d", connectionID))
 	if !ok {
 		return
 	}
@@ -404,7 +405,7 @@ func (r *recording) AddData(connectionID uint32, resource string, id string, fie
 }
 
 func (r *recording) GetData(connectionID uint32, resource string, id string, field string) (*llx.RawData, bool) {
-	asset, ok := r.assets[connectionID]
+	asset, ok := r.assets.Get(fmt.Sprintf("%d", connectionID))
 	if !ok {
 		return nil, false
 	}
@@ -427,7 +428,7 @@ func (r *recording) GetData(connectionID uint32, resource string, id string, fie
 }
 
 func (r *recording) GetResource(connectionID uint32, resource string, id string) (map[string]*llx.RawData, bool) {
-	asset, ok := r.assets[connectionID]
+	asset, ok := r.assets.Get(fmt.Sprintf("%d", connectionID))
 	if !ok {
 		return nil, false
 	}
@@ -475,7 +476,7 @@ func (r *recording) GetAssetRecordings() []*Asset {
 }
 
 func (r *recording) SetAssetRecording(id uint32, reco *Asset) {
-	r.assets[id] = reco
+	r.assets.Set(fmt.Sprintf("%d", id), reco)
 }
 
 // This method makes sure the asset metadata is always included in the data