Fix data race in the topology relation cache

omnibus/config/software/cacerts.rb

@@ -25,13 +25,11 @@
 # doing this.
 name "cacerts"
 
-# We have a synthetic monitor on the latest cacerts file to warn us when the latest
-# cacerts bundle changes.
-# This allows us to always use up-to-date cacerts, without breaking all builds
-# when they change.
-default_version "latest"
-source url: "https://curl.se/ca/cacert.pem",
-       sha256: "f66dff1bdf8f96060b8177976f8b7d9254bc89bc4db933d769f7384d28480bc9",
+# Pinned to a dated curl release: the moving cacert.pem breaks every build when
+# curl publishes a new bundle, and this fork has no monitor for it.
+default_version "2026-09-25"
+source url: "https://curl.se/ca/cacert-#{version}.pem",
+       sha256: "a41b5d356aea97a529fe27e0f7316d2f9d946d75927476cf9cf1b90637d00505",
        target_filename: "cacert.pem"
 
 relative_path "cacerts-#{version}"

pkg/collector/corechecks/cluster/topologycollectors/common.go

@@ -79,7 +79,7 @@ type clusterTopologyCommon struct {
 	possibleRelations             []*topology.Relation
 	k8sVersion                    *version.Info
 	useRelationCache              bool
-	relationCacheWG               sync.WaitGroup
+	relationCacheMu               sync.Mutex
 }
 
 // NewClusterTopologyCommon creates a clusterTopologyCommon
@@ -101,7 +101,6 @@ func NewClusterTopologyCommon(
 		relationChan:                  relationChan,
 		k8sVersion:                    k8sVersion,
 		useRelationCache:              true,
-		relationCacheWG:               sync.WaitGroup{},
 	}
 }
 
@@ -121,18 +120,20 @@ func (c *clusterTopologyCommon) SubmitRelation(relation *topology.Relation) {
 		if sourceExists && targetExists {
 			c.relationChan <- relation
 		} else {
-			c.relationCacheWG.Add(1)
+			c.relationCacheMu.Lock()
 			c.possibleRelations = append(c.possibleRelations, relation)
-			c.relationCacheWG.Done()
+			c.relationCacheMu.Unlock()
 		}
 	} else {
 		c.relationChan <- relation
 	}
 }
 
 func (c *clusterTopologyCommon) CorrelateRelations() {
-	c.relationCacheWG.Add(1)
-	for _, relation := range c.possibleRelations {
+	c.relationCacheMu.Lock()
+	possibleRelations := c.possibleRelations
+	c.relationCacheMu.Unlock()
+	for _, relation := range possibleRelations {
 		_, sourceExists := c.componentIDCache.Load(relation.SourceID)
 		_, targetExists := c.componentIDCache.Load(relation.TargetID)
 		if sourceExists && targetExists {
@@ -145,7 +146,6 @@ func (c *clusterTopologyCommon) CorrelateRelations() {
 			}
 		}
 	}
-	c.relationCacheWG.Done()
 }
 
 // SetUseRelationCache sets if the relation cache should be used or not

pkg/collector/corechecks/cluster/topologycollectors/relation_cache_test.go

@@ -0,0 +1,58 @@
+//go:build kubeapiserver
+
+package topologycollectors
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/StackVista/stackstate-receiver-go-client/pkg/model/topology"
+	"github.com/stretchr/testify/assert"
+)
+
+// Collectors and correlators submit relations from different goroutines; none
+// may be lost while their endpoints are still unknown.
+func TestRelationCacheKeepsConcurrentlySubmittedRelations(t *testing.T) {
+	const submitters, perSubmitter = 8, 200
+	componentChan := make(chan *topology.Component)
+	relationChan := make(chan *topology.Relation)
+	common := NewClusterTopologyCommon(topology.Instance{Type: "kubernetes", URL: "test"}, Kubernetes, nil,
+		componentChan, relationChan, nil)
+
+	var submitted sync.WaitGroup
+	for s := 0; s < submitters; s++ {
+		submitted.Add(1)
+		go func(s int) {
+			defer submitted.Done()
+			for i := 0; i < perSubmitter; i++ {
+				common.SubmitRelation(common.CreateRelation(fmt.Sprintf("source-%d-%d", s, i), "target", "uses"))
+			}
+		}(s)
+	}
+	submitted.Wait()
+
+	go func() {
+		common.SubmitComponent(&topology.Component{ExternalID: "target"})
+		for s := 0; s < submitters; s++ {
+			for i := 0; i < perSubmitter; i++ {
+				common.SubmitComponent(&topology.Component{ExternalID: fmt.Sprintf("source-%d-%d", s, i)})
+			}
+		}
+		common.CorrelateRelations()
+		close(relationChan)
+	}()
+
+	received := 0
+	for {
+		select {
+		case <-componentChan:
+		case _, ok := <-relationChan:
+			if !ok {
+				assert.Equal(t, submitters*perSubmitter, received)
+				return
+			}
+			received++
+		}
+	}
+}