fix: enable race ci and fix problems (#3996)

Enables -race flag for unit tests and FVT tests in run_test_case.yaml to
catch data races in CI.

---------

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

Author: ngjaying

.github/workflows/run_test_case.yaml

@@ -63,13 +63,38 @@ jobs:
           cd extensions/functions/onnx
           sudo ./install.sh
 
+      - name: Install Merge Tool
+        run: go install github.com/wadey/gocovmerge@latest
+
       - name: Run test case
         run: |
           make failpoint-enable
-          go test -trimpath -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")
+          
+          # 1. Run MAIN tests (With Race)
+          # We ADD -skip to avoid running the special tests here
+          go test -trimpath -race -tags="edgex msgpack script parquet test rpc" \
+            -skip "TestMsgpackService|TestExtensions|TestFuncState|TestFuncStateCheckpoint|TestStartCPUProfiling" \
+            -cover -covermode=atomic -coverpkg=./... -coverprofile=cover_race.out \
+            $(go list ./... | grep -v "github.com/lf-edge/ekuiper/v2/fvt")
+      
+          # 2. Run SPECIAL tests (No Race)
+          go test -trimpath -tags="edgex msgpack script parquet test rpc" \
+            -run "TestMsgpackService|TestExtensions|TestFuncState|TestFuncStateCheckpoint|TestStartCPUProfiling" \
+            -cover -covermode=atomic -coverpkg=./... -coverprofile=cover_norace.out \
+            ./internal/service ./internal/topo/exttest ./internal/server/proftest
+      
+          # 3. Safe Merge using gocovmerge
+          # This correctly sums up the counters for shared code
+          $HOME/go/bin/gocovmerge cover_race.out cover_norace.out > coverage.xml
+      
+          # 4. Calculate Total
           total_coverage=$(go tool cover -func=coverage.xml 2>/dev/null | grep total | awk '{print $3}')
-          make failpoint-disable          
           echo "Total coverage: $total_coverage"
+      
+        # 5. Ensure failpoints disable even if tests fail
+      - name: Cleanup Failpoints
+        if: always() # Runs even if previous step failed
+        run: make failpoint-disable
       - uses: actions/upload-artifact@v4
         if: failure()
         with:
@@ -86,7 +111,7 @@ jobs:
       - name: Run fvt
         run: |
           export KUIPER__BASIC__ALLOWEXTERNALFILEACCESS="true"
-          go test -trimpath -tags="full deadlock" --cover -covermode=atomic -coverpkg=./... -coverprofile=fvt_coverage.xml ./fvt
+          go test -trimpath -race -tags="full deadlock" --cover -covermode=atomic -coverpkg=./... -coverprofile=fvt_coverage.xml ./fvt
           total_coverage=$(go tool cover -func=coverage.xml 2>/dev/null | grep total | awk '{print $3}')
           echo "Total coverage: $total_coverage"
       - uses: actions/upload-artifact@v4

extensions/impl/video/source.go

@@ -167,7 +167,9 @@ func (s *Source) runCurrent(ctx api.StreamContext, fps string, ingest api.BytesI
 	// We must read stderr even if DebugResp is false to prevent the ffmpeg pipe from
 	// blocking and to capture the last error message for diagnostics.
 	var lastStderr string
+	stderrDone := make(chan struct{})
 	go func() {
+		defer close(stderrDone)
 		scanner := bufio.NewScanner(stderr)
 		for scanner.Scan() {
 			line := scanner.Text()
@@ -203,6 +205,7 @@ func (s *Source) runCurrent(ctx api.StreamContext, fps string, ingest api.BytesI
 	case <-ctx.Done():
 		return nil
 	case err := <-done:
+		<-stderrDone
 		if err != nil && lastStderr != "" {
 			return fmt.Errorf("%v: %s", err, lastStderr)
 		}

fvt/rule_test.go

@@ -172,7 +172,7 @@ func (s *RuleTestSuite) TestUpsert() {
 		err := server.AddListener(tcp)
 		s.Require().NoError(err)
 		go func() {
-			err = server.Serve()
+			err := server.Serve()
 			fmt.Println(err)
 		}()
 		fmt.Println(tcp.Address())
@@ -270,6 +270,8 @@ func (s *RuleTestSuite) TestUpsert() {
 	})
 	s.Run("compare result", func() {
 		expected := map[string]string{"sim/new1": "{\"b\":2}", "sim/new2": "{\"a\":1}", "sim/old1": "{\"a\":1}", "sim/old2": "{\"b\":2}"}
+		lock.Lock()
+		defer lock.Unlock()
 		s.Require().Equal(expected, result)
 	})
 }

internal/io/http/httpserver/data_server.go

@@ -45,7 +45,10 @@ type GlobalServerManager struct {
 	websocketEndpoint map[string]*websocketEndpointContext
 }
 
-var manager *GlobalServerManager
+var (
+	manager     *GlobalServerManager
+	managerLock syncx.RWMutex
+)
 
 func InitGlobalServerManager(ip string, port int, tlsConf *model.TlsConf) {
 	r := mux.NewRouter()
@@ -66,6 +69,7 @@ func InitGlobalServerManager(ip string, port int, tlsConf *model.TlsConf) {
 			return true
 		},
 	}
+	managerLock.Lock()
 	manager = &GlobalServerManager{
 		websocketEndpoint: map[string]*websocketEndpointContext{},
 		endpoint:          map[string]string{},
@@ -81,20 +85,39 @@ func InitGlobalServerManager(ip string, port int, tlsConf *model.TlsConf) {
 			s.ListenAndServeTLS(conf.Config.Source.HttpServerTls.Certfile, conf.Config.Source.HttpServerTls.Keyfile)
 		}
 	}(manager)
+	managerLock.Unlock()
 	time.Sleep(500 * time.Millisecond)
 }
 
 func ShutDown() {
-	manager.Shutdown()
+	managerLock.RLock()
+	if manager != nil {
+		manager.Shutdown()
+	}
+	managerLock.RUnlock()
+	managerLock.Lock()
 	manager = nil
+	managerLock.Unlock()
 }
 
 func RegisterEndpoint(endpoint string, method string) (string, error) {
-	return manager.RegisterEndpoint(endpoint, method)
+	managerLock.RLock()
+	m := manager
+	managerLock.RUnlock()
+	if m == nil {
+		return "", fmt.Errorf("http server is not running")
+	}
+	return m.RegisterEndpoint(endpoint, method)
 }
 
 func UnregisterEndpoint(endpoint, method string) {
-	manager.UnregisterEndpoint(endpoint, method)
+	managerLock.RLock()
+	m := manager
+	managerLock.RUnlock()
+	if m == nil {
+		return
+	}
+	m.UnregisterEndpoint(endpoint, method)
 }
 
 const (

internal/io/http/httpserver/websocket_server.go

@@ -53,11 +53,23 @@ type websocketEndpointContext struct {
 }
 
 func RegisterWebSocketEndpoint(ctx api.StreamContext, endpoint string) (string, string, error) {
-	return manager.RegisterWebSocketEndpoint(ctx, endpoint)
+	managerLock.RLock()
+	m := manager
+	managerLock.RUnlock()
+	if m == nil {
+		return "", "", fmt.Errorf("http server is not running")
+	}
+	return m.RegisterWebSocketEndpoint(ctx, endpoint)
 }
 
 func UnRegisterWebSocketEndpoint(endpoint string) {
-	wctx := manager.UnRegisterWebSocketEndpoint(endpoint)
+	managerLock.RLock()
+	m := manager
+	managerLock.RUnlock()
+	if m == nil {
+		return
+	}
+	wctx := m.UnRegisterWebSocketEndpoint(endpoint)
 	if wctx != nil {
 		// wait all process exit
 		wctx.wg.Wait()
@@ -144,7 +156,10 @@ func (m *GlobalServerManager) RegisterWebSocketEndpoint(ctx api.StreamContext, e
 		conf.Log.Infof("websocket endpint %v create connection", endpoint)
 	}
 	m.router.HandleFunc(endpoint, func(w http.ResponseWriter, r *http.Request) {
-		if h, ok := m.routes[endpoint]; ok {
+		m.RLock()
+		h, ok := m.routes[endpoint]
+		m.RUnlock()
+		if ok {
 			h(w, r)
 		} else {
 			w.WriteHeader(http.StatusNotFound)

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

internal/io/memory/pubsub/manager.go

@@ -128,13 +128,13 @@ func ProduceError(ctx api.StreamContext, topic string, err error) {
 }
 
 func doProduce(ctx api.StreamContext, topic string, data any) {
+	mu.RLock()
+	defer mu.RUnlock()
 	c, exists := pubTopics[topic]
 	if !exists {
 		return
 	}
 	logger := ctx.GetLogger()
-	mu.RLock()
-	defer mu.RUnlock()
 	// broadcast to all consumers
 	for name, out := range c.consumers {
 		select {

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

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:

internal/io/mqtt/conn.go

@@ -28,9 +28,11 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+	"github.com/lf-edge/ekuiper/v2/pkg/syncx"
 )
 
 type Connection struct {
+	mu syncx.Mutex
 	client.Client
 	id        string
 	server    string
@@ -110,15 +112,20 @@ func (conn *Connection) Status(_ api.StreamContext) modules.ConnectionStatus {
 func (conn *Connection) SetStatusChangeHandler(ctx api.StreamContext, sch api.StatusChangeHandler) {
 	st := conn.status.Load().(modules.ConnectionStatus)
 	sch(st.Status, st.ErrMsg)
+	conn.mu.Lock()
 	conn.scHandler = sch
+	conn.mu.Unlock()
 	ctx.GetLogger().Infof("trigger status change handler")
 }
 
 func (conn *Connection) onConnect(ctx api.StreamContext) {
 	conn.connected.Store(true)
 	conn.status.Store(modules.ConnectionStatus{Status: api.ConnectionConnected})
-	if conn.scHandler != nil {
-		conn.scHandler(api.ConnectionConnected, "")
+	conn.mu.Lock()
+	handler := conn.scHandler
+	conn.mu.Unlock()
+	if handler != nil {
+		handler(api.ConnectionConnected, "")
 	} else {
 		ctx.GetLogger().Warnf("sc handler has not set yet")
 	}
@@ -137,16 +144,22 @@ func (conn *Connection) onConnect(ctx api.StreamContext) {
 func (conn *Connection) onConnectLost(ctx api.StreamContext, err error) {
 	conn.connected.Store(false)
 	conn.status.Store(modules.ConnectionStatus{Status: api.ConnectionDisconnected, ErrMsg: err.Error()})
-	if conn.scHandler != nil {
-		conn.scHandler(api.ConnectionDisconnected, err.Error())
+	conn.mu.Lock()
+	handler := conn.scHandler
+	conn.mu.Unlock()
+	if handler != nil {
+		handler(api.ConnectionDisconnected, err.Error())
 	}
 	ctx.GetLogger().Infof("%v", err)
 }
 
 func (conn *Connection) onReconnecting(ctx api.StreamContext) {
 	conn.status.Store(modules.ConnectionStatus{Status: api.ConnectionConnecting})
-	if conn.scHandler != nil {
-		conn.scHandler(api.ConnectionConnecting, "")
+	conn.mu.Lock()
+	handler := conn.scHandler
+	conn.mu.Unlock()
+	if handler != nil {
+		handler(api.ConnectionConnecting, "")
 	}
 	ctx.GetLogger().Debugf("Reconnecting to mqtt broker")
 }