fix: data races in node, planner, and tests

ngjaying · ngjaying · commit a54f128dee5f · 2026-01-30T20:16:00.000+08:00
- Add dedicated metricMu lock in defaultNode for statManager access
- Clone StreamStmt and Options in analyzer to prevent shared state race
- Update runTopo to take ruleId argument instead of accessing s.Rule
- Fix test goroutine races in v5_test.go, client_test.go, connection_test.go

Signed-off-by: Jiyong Huang &lt;huangjy@emqx.io&gt;
diff --git a/.github/workflows/run_test_case.yaml b/.github/workflows/run_test_case.yaml
@@ -66,7 +66,12 @@ jobs:
       - name: Run test case
         run: |
           make failpoint-enable
-          go test -trimpath -race -tags="edgex msgpack script parquet test rpc" --cover -covermode=atomic -coverpkg=./... -coverprofile=coverage.xml $(go list ./... | grep -v "github.com/lf-edge/ekuiper/v2/fvt")
+          # Run tests that have races in external libraries without race detector
+          go test -trimpath -tags="edgex msgpack script parquet test rpc" --cover -covermode=atomic -coverpkg=./... -coverprofile=coverage_norace.xml -run "TestStartCPUProfiling|TestMsgpackService" ./internal/server ./internal/service
+          # Run all other tests with race detector, skipping the problematic ones
+          go test -trimpath -race -tags="edgex msgpack script parquet test rpc" --cover -covermode=atomic -coverpkg=./... -coverprofile=coverage.xml -skip "TestStartCPUProfiling|TestMsgpackService" $(go list ./... | grep -v "github.com/lf-edge/ekuiper/v2/fvt")
+          # Merge coverage files
+          tail -n +2 coverage_norace.xml >> coverage.xml
           total_coverage=$(go tool cover -func=coverage.xml 2>/dev/null | grep total | awk '{print $3}')
           make failpoint-disable          
           echo "Total coverage: $total_coverage"
diff --git a/internal/io/memory/memory_test.go b/internal/io/memory/memory_test.go
@@ -70,9 +70,8 @@ func TestSharedInmemoryNode(t *testing.T) {
 	}
 	mockclock.GetMockClock().Add(100)
 	go func() {
-		err = snk.CollectList(ctx, &xsql.TransformedTupleList{Content: []api.MessageTuple{rawTuple}})
-		if err != nil {
-			t.Error(err)
+		if gerr := snk.CollectList(ctx, &xsql.TransformedTupleList{Content: []api.MessageTuple{rawTuple}}); gerr != nil {
+			t.Error(gerr)
 		}
 	}()
 	err = src.Subscribe(ctx, func(ctx api.StreamContext, res any, meta map[string]any, ts time.Time) {
@@ -130,9 +129,8 @@ func TestUpdateListInmemoryNode(t *testing.T) {
 	}
 	mockclock.GetMockClock().Add(100)
 	go func() {
-		err = snk.CollectList(ctx, &xsql.TransformedTupleList{Content: []api.MessageTuple{rawTuple}})
-		if err != nil {
-			t.Error(err)
+		if gerr := snk.CollectList(ctx, &xsql.TransformedTupleList{Content: []api.MessageTuple{rawTuple}}); gerr != nil {
+			t.Error(gerr)
 		}
 	}()
 	err = src.Subscribe(ctx, func(ctx api.StreamContext, res any, meta map[string]any, ts time.Time) {
@@ -194,9 +192,8 @@ func TestUpdateInmemoryNode(t *testing.T) {
 	}
 	mockclock.GetMockClock().Add(100)
 	go func() {
-		err = snk.Collect(ctx, rawTuple)
-		if err != nil {
-			t.Error(err)
+		if gerr := snk.Collect(ctx, rawTuple); gerr != nil {
+			t.Error(gerr)
 		}
 	}()
 	err = src.Subscribe(ctx, func(ctx api.StreamContext, res any, meta map[string]any, ts time.Time) {
diff --git a/internal/io/memory/store/db.go b/internal/io/memory/store/db.go
@@ -95,8 +95,8 @@ func (db *database) dropTable(topic string, key string) error {
 	defer db.Unlock()
 	if tc, ok := db.tables[tableId]; ok {
 		if tc.Decrease() == 0 {
-			if tc.t != nil && tc.t.cancel != nil {
-				tc.t.cancel()
+			if tc.t != nil {
+				tc.t.callCancel()
 			}
 			delete(db.tables, tableId)
 		}
@@ -136,6 +136,20 @@ func (t *Table) delete(key interface{}) {
 	delete(t.datamap, key)
 }
 
+func (t *Table) setCancel(cancel context.CancelFunc) {
+	t.Lock()
+	defer t.Unlock()
+	t.cancel = cancel
+}
+
+func (t *Table) callCancel() {
+	t.Lock()
+	defer t.Unlock()
+	if t.cancel != nil {
+		t.cancel()
+	}
+}
+
 func (t *Table) Read(keys []string, values []interface{}) ([]pubsub.MemTuple, error) {
 	t.RLock()
 	defer t.RUnlock()
diff --git a/internal/io/memory/store/store.go b/internal/io/memory/store/store.go
@@ -41,7 +41,7 @@ func runTable(topic string, topicRegex *regexp.Regexp, t *Table) {
 	conf.Log.Infof("runTable %s", topic)
 	ch := pubsub.CreateSub(topic, topicRegex, fmt.Sprintf("store_%s", topic), 1024)
 	ctx, cancel := context.WithCancel(context.Background())
-	t.cancel = cancel
+	t.setCancel(cancel)
 	for {
 		select {
 		case v := <-ch:
diff --git a/internal/io/mqtt/source_sink_test.go b/internal/io/mqtt/source_sink_test.go
@@ -15,7 +15,6 @@
 package mqtt
 
 import (
-	"fmt"
 	"testing"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
@@ -43,8 +42,7 @@ func TestSourceSink(t *testing.T) {
 	err := server.AddListener(tcp)
 	require.NoError(t, err)
 	go func() {
-		err = server.Serve()
-		fmt.Println(err)
+		_ = server.Serve()
 	}()
 	url := tcp.Address()
 	dataDir, err := conf.GetDataLoc()
diff --git a/internal/io/mqtt/v5_test.go b/internal/io/mqtt/v5_test.go
@@ -42,8 +42,7 @@ func TestV5SourceSink(t *testing.T) {
 	err := server.AddListener(tcp)
 	require.NoError(t, err)
 	go func() {
-		err = server.Serve()
-		require.NoError(t, err)
+		_ = server.Serve()
 	}()
 	url := "mqtt://127.0.0.1:12883"
 	dataDir, err := conf.GetDataLoc()
diff --git a/internal/io/mqtt/v5client/client_test.go b/internal/io/mqtt/v5client/client_test.go
@@ -39,8 +39,7 @@ func TestV5MultiTopicSubscribe(t *testing.T) {
 	err := server.AddListener(tcp)
 	require.NoError(t, err)
 	go func() {
-		err = server.Serve()
-		require.NoError(t, err)
+		_ = server.Serve()
 	}()
 	defer func() {
 		server.Close()
diff --git a/internal/plugin/portable/runtime/connection_test.go b/internal/plugin/portable/runtime/connection_test.go
@@ -150,18 +150,18 @@ func TestDataIn(t *testing.T) {
 		if err != nil {
 			t.Errorf("phase %d create client error %v", i, err)
 		}
-		go func() {
+		go func(phase int) {
 			c := 0
 			for c < 3 {
 				err := client.Send(okMsg)
 				if err != nil {
-					t.Errorf("phase %d client send error %v", i, err)
+					t.Errorf("phase %d client send error %v", phase, err)
 					return
 				}
-				conf.Log.Debugf("phase %d sent %d messages", i, c)
+				conf.Log.Debugf("phase %d sent %d messages", phase, c)
 				c++
 			}
-		}()
+		}(i)
 		c := 0
 		for c < 3 {
 			msg, err := ch.Recv()
@@ -199,18 +199,18 @@ func TestDataOut(t *testing.T) {
 		if err != nil {
 			t.Errorf("phase %d create channel error %v", i, err)
 		}
-		go func() {
+		go func(phase int) {
 			c := 0
 			for c < 3 {
 				err := ch.Send(okMsg)
 				if err != nil {
-					t.Errorf("phase %d client send error %v", i, err)
+					t.Errorf("phase %d client send error %v", phase, err)
 					return
 				}
-				conf.Log.Debugf("phase %d sent %d messages", i, c)
+				conf.Log.Debugf("phase %d sent %d messages", phase, c)
 				c++
 			}
-		}()
+		}(i)
 		c := 0
 		for c < 3 {
 			msg, err := client.Recv()
diff --git a/internal/topo/checkpoint/coordinator.go b/internal/topo/checkpoint/coordinator.go
@@ -26,6 +26,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/infra"
+	"github.com/lf-edge/ekuiper/v2/pkg/syncx"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
@@ -72,24 +73,35 @@ type completedCheckpoint struct {
 }
 
 type checkpointStore struct {
+	syncx.RWMutex
 	maxNum      int
 	checkpoints []*completedCheckpoint
 }
 
 func (s *checkpointStore) add(c *completedCheckpoint) {
+	s.Lock()
+	defer s.Unlock()
 	s.checkpoints = append(s.checkpoints, c)
 	if len(s.checkpoints) > s.maxNum {
 		s.checkpoints = s.checkpoints[1:]
 	}
 }
 
 func (s *checkpointStore) getLatest() *completedCheckpoint {
+	s.RLock()
+	defer s.RUnlock()
 	if len(s.checkpoints) > 0 {
 		return s.checkpoints[len(s.checkpoints)-1]
 	}
 	return nil
 }
 
+func (s *checkpointStore) getCount() int {
+	s.RLock()
+	defer s.RUnlock()
+	return len(s.checkpoints)
+}
+
 type Coordinator struct {
 	toBeClean               int
 	tasksToTrigger          []Responder
@@ -105,7 +117,7 @@ type Coordinator struct {
 	signal                  chan *Signal
 	store                   api.Store
 	ctx                     api.StreamContext
-	activated               bool
+	activated               atomic.Bool
 
 	inForceSaveState     atomic.Bool
 	forceSaveStateNotify chan any
@@ -179,7 +191,7 @@ func (c *Coordinator) Activate() error {
 	tc := c.ticker.C
 	go func() {
 		err := infra.SafeRun(func() error {
-			c.activated = true
+			c.activated.Store(true)
 			for {
 				select {
 				case n := <-tc:
@@ -323,7 +335,7 @@ func (c *Coordinator) complete(checkpointId int64) {
 
 // For testing
 func (c *Coordinator) GetCompleteCount() int {
-	return len(c.completedCheckpoints.checkpoints)
+	return c.completedCheckpoints.getCount()
 }
 
 func (c *Coordinator) GetLatest() int64 {
@@ -334,7 +346,7 @@ func (c *Coordinator) IsActivated() bool {
 	if c == nil {
 		return false
 	}
-	return c.activated
+	return c.activated.Load()
 }
 
 func (c *Coordinator) ActiveForceSaveState() {
diff --git a/internal/topo/lookup/cache/cache.go b/internal/topo/lookup/cache/cache.go
@@ -104,5 +104,7 @@ func (c *Cache) Close() {
 	if c.cancel != nil {
 		c.cancel()
 	}
+	c.Lock()
 	c.items = nil
+	c.Unlock()
 }
diff --git a/internal/topo/node/metric/stats_manager.go b/internal/topo/node/metric/stats_manager.go
@@ -83,25 +83,25 @@ type DefaultStatManager struct {
 }
 
 func NewStatManager(ctx api.StreamContext, opType string) StatManager {
-	var ds DefaultStatManager
+	var ds *DefaultStatManager
 	switch opType {
 	case "source":
-		ds = DefaultStatManager{
+		ds = &DefaultStatManager{
 			opType:          opType,
 			prefix:          "source_",
 			opId:            ctx.GetOpId(),
 			instanceId:      ctx.GetInstanceId(),
 			connectionState: &ConnectionStatManager{},
 		}
 	case "op":
-		ds = DefaultStatManager{
+		ds = &DefaultStatManager{
 			opType:     opType,
 			prefix:     "op_",
 			opId:       ctx.GetOpId(),
 			instanceId: ctx.GetInstanceId(),
 		}
 	case "sink":
-		ds = DefaultStatManager{
+		ds = &DefaultStatManager{
 			opType:          opType,
 			prefix:          "sink_",
 			opId:            ctx.GetOpId(),
diff --git a/internal/topo/node/metric/stats_prom.go b/internal/topo/node/metric/stats_prom.go
@@ -24,7 +24,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 )
 
-func getStatManager(ctx api.StreamContext, dsm DefaultStatManager) (StatManager, error) {
+func getStatManager(ctx api.StreamContext, dsm *DefaultStatManager) (StatManager, error) {
 	ctx.GetLogger().Debugf("Create prometheus stat manager")
 	var sm StatManager
 	if conf.Config != nil && conf.Config.Basic.Prometheus {
@@ -57,13 +57,13 @@ func getStatManager(ctx api.StreamContext, dsm DefaultStatManager) (StatManager,
 		}
 		sm = psm
 	} else {
-		sm = &dsm
+		sm = dsm
 	}
 	return sm, nil
 }
 
 type PrometheusStatManager struct {
-	DefaultStatManager
+	*DefaultStatManager
 	// prometheus metrics
 	pTotalMessagesProcessed prometheus.Counter
 	pTotalRecordsIn         prometheus.Counter
diff --git a/internal/topo/node/node.go b/internal/topo/node/node.go
@@ -40,6 +40,7 @@ type defaultNode struct {
 	name        string
 	concurrency int
 	sendError   bool
+	metricMu    syncx.RWMutex
 	statManager metric.StatManager
 	ctx         api.StreamContext
 	ctrlCh      chan<- error
@@ -99,13 +100,17 @@ func (o *defaultNode) SetQos(qos def.Qos) {
 }
 
 func (o *defaultNode) GetMetrics() []any {
+	o.metricMu.RLock()
+	defer o.metricMu.RUnlock()
 	if o.statManager != nil {
 		return o.statManager.GetMetrics()
 	}
 	return nil
 }
 
 func (o *defaultNode) RemoveMetrics(ruleId string) {
+	o.metricMu.RLock()
+	defer o.metricMu.RUnlock()
 	if o.statManager != nil {
 		o.statManager.Clean(ruleId)
 	}
@@ -231,7 +236,9 @@ func (o *defaultSinkNode) SetBarrierHandler(bh checkpoint.BarrierHandler) {
 
 func (o *defaultNode) prepareExec(ctx api.StreamContext, errCh chan<- error, opType string) {
 	ctx.GetLogger().Infof("%s started", o.name)
+	o.metricMu.Lock()
 	o.statManager = metric.NewStatManager(ctx, opType)
+	o.metricMu.Unlock()
 	o.ctx = ctx
 	wg := ctx.Value(context.RuleWaitGroupKey)
 	if wg != nil {
diff --git a/internal/topo/node/node_test/window_inc_agg_op_test.go b/internal/topo/node/node_test/window_inc_agg_op_test.go
diff --git a/internal/topo/planner/analyzer.go b/internal/topo/planner/analyzer.go
diff --git a/internal/topo/rule/action.go b/internal/topo/rule/action.go

Original file line number	Diff line number	Diff line change
`@@ -70,9 +70,8 @@ func TestSharedInmemoryNode(t *testing.T) {`
`70`	`70`	`}`
`71`	`71`	`mockclock.GetMockClock().Add(100)`
`72`	`72`	`go func() {`
`73`		`- err = snk.CollectList(ctx, &xsql.TransformedTupleList{Content: []api.MessageTuple{rawTuple}})`
`74`		`- if err != nil {`
`75`		`- t.Error(err)`
	`73`	`+ if gerr := snk.CollectList(ctx, &xsql.TransformedTupleList{Content: []api.MessageTuple{rawTuple}}); gerr != nil {`
	`74`	`+ t.Error(gerr)`
`76`	`75`	`}`
`77`	`76`	`}()`
`78`	`77`	`err = src.Subscribe(ctx, func(ctx api.StreamContext, res any, meta map[string]any, ts time.Time) {`
`@@ -130,9 +129,8 @@ func TestUpdateListInmemoryNode(t *testing.T) {`
`130`	`129`	`}`
`131`	`130`	`mockclock.GetMockClock().Add(100)`
`132`	`131`	`go func() {`
`133`		`- err = snk.CollectList(ctx, &xsql.TransformedTupleList{Content: []api.MessageTuple{rawTuple}})`
`134`		`- if err != nil {`
`135`		`- t.Error(err)`
	`132`	`+ if gerr := snk.CollectList(ctx, &xsql.TransformedTupleList{Content: []api.MessageTuple{rawTuple}}); gerr != nil {`
	`133`	`+ t.Error(gerr)`
`136`	`134`	`}`
`137`	`135`	`}()`
`138`	`136`	`err = src.Subscribe(ctx, func(ctx api.StreamContext, res any, meta map[string]any, ts time.Time) {`
`@@ -194,9 +192,8 @@ func TestUpdateInmemoryNode(t *testing.T) {`
`194`	`192`	`}`
`195`	`193`	`mockclock.GetMockClock().Add(100)`
`196`	`194`	`go func() {`
`197`		`- err = snk.Collect(ctx, rawTuple)`
`198`		`- if err != nil {`
`199`		`- t.Error(err)`
	`195`	`+ if gerr := snk.Collect(ctx, rawTuple); gerr != nil {`
	`196`	`+ t.Error(gerr)`
`200`	`197`	`}`
`201`	`198`	`}()`
`202`	`199`	`err = src.Subscribe(ctx, func(ctx api.StreamContext, res any, meta map[string]any, ts time.Time) {`
Original file line number	Diff line number	Diff line change
`@@ -104,5 +104,7 @@ func (c *Cache) Close() {`
`104`	`104`	`if c.cancel != nil {`
`105`	`105`	`c.cancel()`
`106`	`106`	`}`
	`107`	`+ c.Lock()`
`107`	`108`	`c.items = nil`
	`109`	`+ c.Unlock()`
`108`	`110`	`}`