test(stability): add query benchmarks + a concurrent read/write test

Benchmarks (internal/graphstore, internal/query) for the hot read paths —
QueryEntities, QueryTopo, and the full Service.Execute entity path — so perf
regressions on large result sets become visible. 'make bench' runs them (smoke
by default, BENCHTIME-overridable); CI runs them non-blocking.

A concurrent read/write test stresses the memory store's RWMutex; under
'go test -race' it is a data-race regression guard for the shared maps
(verified race-clean).

Author: qiansheng91

.github/workflows/ci.yml

@@ -78,6 +78,12 @@ jobs:
       - name: Run CI
         run: make ci
 
+      # Smoke-run the benchmarks so they keep compiling and executing. Non-blocking
+      # for now (no perf thresholds yet); drop continue-on-error to gate on perf.
+      - name: Run benchmarks (smoke)
+        run: make bench
+        continue-on-error: true
+
       - name: Run Playwright UI tests
         run: |
           make quickstart &

Makefile

@@ -1,5 +1,5 @@
 .PHONY: help check-env install-env setup setup-ui expand doc docs-schema docs-schema-check example-validate check-manifest
-.PHONY: build build-service build-cli install-cli build-ui build-sdk-go dev quickstart dev-api dev-web deploy serve-ui status stop-all stop-dev stop-deploy test test-service test-ui test-ui-e2e test-capability test-quickstart-health test-ladybug verify verify-go verify-python verify-java guard ci clean
+.PHONY: build build-service build-cli install-cli build-ui build-sdk-go dev quickstart dev-api dev-web deploy serve-ui status stop-all stop-dev stop-deploy test test-service bench test-ui test-ui-e2e test-capability test-quickstart-health test-ladybug verify verify-go verify-python verify-java guard ci clean
 
 VENV_PYTHON := .venv/bin/python
 CONDA_PYTHON := $(if $(CONDA_PREFIX),$(CONDA_PREFIX)/bin/python)
@@ -135,6 +135,13 @@ serve-ui: build-ui
 test-service:
 	go test ./...
 
+# Run benchmarks. Default is a smoke run (a few iterations: ensures they compile
+# and execute without panicking). Raise BENCHTIME, e.g. `make bench BENCHTIME=2s`,
+# for real measurements.
+BENCHTIME ?= 10x
+bench:
+	go test -run '^$$' -bench=. -benchtime=$(BENCHTIME) ./internal/query/... ./internal/graphstore/...
+
 test-ui:
 	@PNPM="$(PNPM)" bash ./scripts/env.sh web-build
 

internal/graphstore/bench_test.go

@@ -0,0 +1,61 @@
+package graphstore
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/alibaba/UnifiedModel/pkg/model"
+)
+
+// seedMemoryStore fills a workspace with n entities and n relations (a ring) so
+// the scan paths have realistic work to do.
+func seedMemoryStore(tb testing.TB, store *MemoryStore, n int) {
+	tb.Helper()
+	ctx := context.Background()
+	ents := make([]model.EntityPayload, n)
+	rels := make([]model.RelationPayload, n)
+	for i := 0; i < n; i++ {
+		id := fmt.Sprintf("%032x", i+1)
+		ents[i] = entityPayload(id, "Update", 100, 200, map[string]any{"display_name": fmt.Sprintf("svc-%d", i)})
+	}
+	if _, err := store.WriteEntities(ctx, model.EntityWriteBatch{Workspace: "demo", Entities: ents}); err != nil {
+		tb.Fatalf("seed entities: %v", err)
+	}
+	for i := 0; i < n; i++ {
+		src := fmt.Sprintf("%032x", i+1)
+		dst := fmt.Sprintf("%032x", (i+1)%n+1)
+		rels[i] = relationPayload(src, dst, "Update", 100, 200, nil)
+	}
+	if _, err := store.WriteRelations(ctx, model.RelationWriteBatch{Workspace: "demo", Relations: rels}); err != nil {
+		tb.Fatalf("seed relations: %v", err)
+	}
+}
+
+func BenchmarkQueryEntities(b *testing.B) {
+	store := NewMemoryStore()
+	seedMemoryStore(b, store, 2000)
+	ctx := context.Background()
+	plan := model.EntityQueryPlan{Workspace: "demo", Limit: 1000}
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		if _, err := store.QueryEntities(ctx, plan); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkQueryTopo(b *testing.B) {
+	store := NewMemoryStore()
+	seedMemoryStore(b, store, 2000)
+	ctx := context.Background()
+	plan := model.TopoQueryPlan{Workspace: "demo", Limit: 1000}
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		if _, err := store.QueryTopo(ctx, plan); err != nil {
+			b.Fatal(err)
+		}
+	}
+}

internal/graphstore/concurrent_test.go

@@ -0,0 +1,68 @@
+package graphstore
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/alibaba/UnifiedModel/pkg/model"
+)
+
+// TestConcurrentReadWrite stresses the memory store's RWMutex with overlapping
+// readers and writers. On its own it asserts no operation errors; under `go test
+// -race` (the gate's mode) it is a data-race regression guard for the shared
+// entity/relation maps.
+func TestConcurrentReadWrite(t *testing.T) {
+	store := NewMemoryStore()
+	seedMemoryStore(t, store, 200)
+
+	const (
+		readers = 24
+		writers = 8
+		iters   = 100
+	)
+	errCh := make(chan error, readers+writers)
+	var wg sync.WaitGroup
+
+	for r := 0; r < readers; r++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			ctx := context.Background()
+			for i := 0; i < iters; i++ {
+				if _, err := store.QueryEntities(ctx, model.EntityQueryPlan{Workspace: "demo", Limit: 100}); err != nil {
+					errCh <- fmt.Errorf("query entities: %w", err)
+					return
+				}
+				if _, err := store.QueryTopo(ctx, model.TopoQueryPlan{Workspace: "demo", Limit: 100}); err != nil {
+					errCh <- fmt.Errorf("query topo: %w", err)
+					return
+				}
+			}
+		}()
+	}
+	for w := 0; w < writers; w++ {
+		wg.Add(1)
+		go func(w int) {
+			defer wg.Done()
+			ctx := context.Background()
+			for i := 0; i < iters; i++ {
+				id := fmt.Sprintf("%032x", 1_000_000+w*iters+i)
+				if _, err := store.WriteEntities(ctx, model.EntityWriteBatch{
+					Workspace: "demo",
+					Entities:  []model.EntityPayload{entityPayload(id, "Update", 100, 200, nil)},
+				}); err != nil {
+					errCh <- fmt.Errorf("write entities: %w", err)
+					return
+				}
+			}
+		}(w)
+	}
+
+	wg.Wait()
+	close(errCh)
+	for err := range errCh {
+		t.Fatalf("concurrent operation failed: %v", err)
+	}
+}

internal/query/bench_test.go

@@ -0,0 +1,47 @@
+package query
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/alibaba/UnifiedModel/internal/graphstore"
+	"github.com/alibaba/UnifiedModel/pkg/model"
+)
+
+// BenchmarkExecuteEntityQuery measures the full read entry point: normalize ->
+// parse -> plan -> executor -> store scan, returning matched rows.
+func BenchmarkExecuteEntityQuery(b *testing.B) {
+	store := graphstore.NewMemoryStore()
+	ctx := context.Background()
+	ents := make([]model.EntityPayload, 1000)
+	for i := range ents {
+		ents[i] = model.EntityPayload{
+			"__domain__":              "devops",
+			"__entity_type__":         "devops.service",
+			"__entity_id__":           fmt.Sprintf("%032x", i+1),
+			"__method__":              "Update",
+			"__first_observed_time__": int64(100),
+			"__last_observed_time__":  int64(200),
+			"__keep_alive_seconds__":  int64(60),
+			"display_name":            fmt.Sprintf("svc-%d", i),
+		}
+	}
+	if _, err := store.WriteEntities(ctx, model.EntityWriteBatch{Workspace: "demo", Entities: ents}); err != nil {
+		b.Fatalf("seed: %v", err)
+	}
+
+	svc := NewService(store)
+	req := model.QueryRequest{Query: ".entity with(domain='devops', name='devops.service') | limit 1000"}
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		res, err := svc.Execute(ctx, "demo", req)
+		if err != nil {
+			b.Fatal(err)
+		}
+		if len(res.Rows) == 0 {
+			b.Fatal("expected rows")
+		}
+	}
+}