ci: split race tests and merge coverage

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 -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")
+          
+          # 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/server ./internal/service ./internal/topo/topotest
+      
+          # 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:

internal/io/http/httpserver/data_server.go

@@ -45,8 +45,10 @@ type GlobalServerManager struct {
 	websocketEndpoint map[string]*websocketEndpointContext
 }
 
-var manager *GlobalServerManager
-var managerLock sync.RWMutex
+var (
+	manager     *GlobalServerManager
+	managerLock syncx.RWMutex
+)
 
 func InitGlobalServerManager(ip string, port int, tlsConf *model.TlsConf) {
 	r := mux.NewRouter()
@@ -100,20 +102,22 @@ func ShutDown() {
 
 func RegisterEndpoint(endpoint string, method string) (string, error) {
 	managerLock.RLock()
-	defer managerLock.RUnlock()
-	if manager == nil {
+	m := manager
+	managerLock.RUnlock()
+	if m == nil {
 		return "", fmt.Errorf("http server is not running")
 	}
-	return manager.RegisterEndpoint(endpoint, method)
+	return m.RegisterEndpoint(endpoint, method)
 }
 
 func UnregisterEndpoint(endpoint, method string) {
 	managerLock.RLock()
-	defer managerLock.RUnlock()
-	if manager == nil {
+	m := manager
+	managerLock.RUnlock()
+	if m == nil {
 		return
 	}
-	manager.UnregisterEndpoint(endpoint, method)
+	m.UnregisterEndpoint(endpoint, method)
 }
 
 const (

internal/topo/node/node_test/window_inc_agg_op_test.go

@@ -525,7 +525,7 @@ func TestIncAggSlidingWindowDelay(t *testing.T) {
 }
 
 func waitExecute() {
-	time.Sleep(100 * time.Millisecond)
+	time.Sleep(500 * time.Millisecond)
 }
 
 func TestIncHoppingWindow(t *testing.T) {

internal/topo/topotest/mocknode/mock_source.go

@@ -76,8 +76,8 @@ func (m *MockSource) Subscribe(ctx api.StreamContext, ingest api.TupleIngest, in
 			m.Lock()
 			m.offset = i + 1
 			m.Unlock()
-			ingest(ctx, map[string]any(d.Message), map[string]any{"topic": "mock"}, timex.GetNow())
 			log.Debugf("%d: mock source %s is sending data %d:%v", timex.GetNowInMilli(), ctx.GetOpId(), i, d)
+			ingest(ctx, map[string]any(d.Message), map[string]any{"topic": "mock"}, timex.GetNow())
 		case <-ctx.Done():
 			log.Debugf("mock source open DONE")
 			return nil