fix: data races in node, planner, and tests

- 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 <huangjy@emqx.io>

Author: ngjaying

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()

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()

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()

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 {

internal/topo/planner/analyzer.go

@@ -521,8 +521,15 @@ func convertStreamInfo(streamStmt *ast.StreamStmt) (*streamInfo, error) {
 			return nil, err
 		}
 	}
+	// Clone the statement to avoid data race when multiple rules sharing the same stream definition
+	// The planner may modify the options, e.g. set the type to default value
+	newStmt := *streamStmt
+	if streamStmt.Options != nil {
+		newOpt := *streamStmt.Options
+		newStmt.Options = &newOpt
+	}
 	return &streamInfo{
-		stmt:   streamStmt,
+		stmt:   &newStmt,
 		schema: ss,
 	}, nil
 }

internal/topo/rule/action.go

@@ -88,7 +88,7 @@ func (s *State) doStart() error {
 				s.topoGraph = s.topology.GetTopo()
 			}
 		}
-		go s.runTopo(s.topology)
+		go s.runTopo(s.topology, s.Rule.Id)
 		return nil
 	})
 	if err != nil {
@@ -116,7 +116,7 @@ func (s *State) doStop(stateType machine.RunState, msg string) {
 }
 
 // This is called async
-func (s *State) runTopo(tp *topo.Topo) {
+func (s *State) runTopo(tp *topo.Topo, ruleId string) {
 	s.logger.Infof("topo %d opens", tp.GetRunId())
 	e := <-tp.Open()
 	s.logger.Infof("topo %d stops", tp.GetRunId())
@@ -135,7 +135,7 @@ func (s *State) runTopo(tp *topo.Topo) {
 			if len(msg) > 0 {
 				lastWill = fmt.Sprintf("%s: %s", lastWill, msg)
 			}
-			s.updateTrigger(s.Rule.Id, false)
+			s.updateTrigger(ruleId, false)
 		}
 	}
 	// The run exit may be caused by user action or rule itself