perf(graphrag,mcp,storage,ui): scale topology to 100-200 services + labelled, searchable map

internal/config/config.go

@@ -185,11 +185,12 @@ type Config struct {
 	IngestPipelineWorkers  int // default 8 worker goroutines
 	// IngestPipelinePerTenantCap caps in-flight batches per tenant so a noisy
 	// tenant cannot starve siblings of fresh queue slots when fullness is
-	// below the soft-backpressure threshold. 0 (default) disables — single-
-	// tenant deployments need no cap. Operators on multi-tenant deployments
-	// should set INGEST_PIPELINE_PER_TENANT_CAP to roughly Capacity/N where
-	// N is the expected number of concurrently-active tenants, with some
-	// headroom (e.g. 2× the fair-share value) for short bursts.
+	// below the soft-backpressure threshold. When unset it defaults to ~30% of
+	// the resolved queue size (see Load) so multi-tenant deployments are
+	// protected out of the box; an explicit INGEST_PIPELINE_PER_TENANT_CAP=0
+	// disables the cap for single-tenant deployments. Operators can instead
+	// pin it to roughly Capacity/N where N is the expected number of
+	// concurrently-active tenants, with headroom for short bursts.
 	IngestPipelinePerTenantCap int
 
 	// TLS (HTTP + gRPC). When both paths are set, TLS is enabled on both servers.
@@ -377,6 +378,16 @@ func Load(customPath string) (*Config, error) {
 		AllowSqliteProd: parseTruthy(getEnv("OTELCONTEXT_ALLOW_SQLITE_PROD", "")),
 	}
 	applyDriverDefaults(cfg)
+
+	// Derive a sane per-tenant ingest cap when the operator did not set one.
+	// Run AFTER applyDriverDefaults so it tracks the (possibly SQLite-adjusted)
+	// queue size: ~30% of the queue lets a single tenant burst but stops one
+	// noisy tenant from monopolising every slot at 100–200 services. An explicit
+	// INGEST_PIPELINE_PER_TENANT_CAP=0 is respected as "disabled".
+	if _, set := os.LookupEnv("INGEST_PIPELINE_PER_TENANT_CAP"); !set && cfg.IngestPipelinePerTenantCap == 0 {
+		cfg.IngestPipelinePerTenantCap = cfg.IngestPipelineQueueSize * 30 / 100
+	}
+
 	return cfg, nil
 }
 

internal/graphrag/adjacency_index_test.go

@@ -0,0 +1,169 @@
+package graphrag
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"testing"
+	"time"
+)
+
+// edgeIDs collects from→to keys so two []*Edge slices can be compared
+// order-independently.
+func edgeIDs(edges []*Edge) []string {
+	out := make([]string, 0, len(edges))
+	for _, e := range edges {
+		out = append(out, e.FromID+"->"+e.ToID)
+	}
+	sort.Strings(out)
+	return out
+}
+
+func fullScanFrom(s *ServiceStore, service string) []*Edge {
+	var out []*Edge
+	for _, e := range s.Edges {
+		if e.Type == EdgeCalls && e.FromID == service {
+			out = append(out, e)
+		}
+	}
+	return out
+}
+
+func fullScanTo(s *ServiceStore, service string) []*Edge {
+	var out []*Edge
+	for _, e := range s.Edges {
+		if e.Type == EdgeCalls && e.ToID == service {
+			out = append(out, e)
+		}
+	}
+	return out
+}
+
+// TestAdjacencyIndexMatchesFullScan asserts the indexed CallEdgesFrom/To return
+// exactly the same CALLS edges a full Edges-map scan would, across both the
+// aggregating UpsertCallEdge path and the existence-only EnsureCallEdge path.
+func TestAdjacencyIndexMatchesFullScan(t *testing.T) {
+	g := newTestGraphRAG(t)
+	now := time.Now()
+	s := g.storesForTenant("t").service
+
+	// Aggregating path.
+	s.UpsertCallEdge("a", "b", 5, false, now)
+	s.UpsertCallEdge("a", "c", 5, true, now)
+	s.UpsertCallEdge("b", "c", 5, false, now)
+	// Existence-only (pre-sample topology observer) path.
+	s.EnsureService("d", now)
+	s.EnsureCallEdge("a", "d", now)
+	s.EnsureCallEdge("d", "c", now)
+	// Repeat upserts must not duplicate index entries.
+	s.UpsertCallEdge("a", "b", 5, false, now)
+	s.EnsureCallEdge("a", "d", now)
+
+	for _, svc := range []string{"a", "b", "c", "d"} {
+		gotFrom := edgeIDs(s.CallEdgesFrom(svc))
+		wantFrom := edgeIDs(fullScanFrom(s, svc))
+		if fmt.Sprint(gotFrom) != fmt.Sprint(wantFrom) {
+			t.Errorf("CallEdgesFrom(%q) = %v, full scan = %v", svc, gotFrom, wantFrom)
+		}
+		gotTo := edgeIDs(s.CallEdgesTo(svc))
+		wantTo := edgeIDs(fullScanTo(s, svc))
+		if fmt.Sprint(gotTo) != fmt.Sprint(wantTo) {
+			t.Errorf("CallEdgesTo(%q) = %v, full scan = %v", svc, gotTo, wantTo)
+		}
+	}
+
+	// a→b should be a single edge despite the repeated upsert.
+	if from := s.CallEdgesFrom("a"); len(from) != 3 { // a→b, a→c, a→d
+		t.Fatalf("CallEdgesFrom(a) = %d edges, want 3", len(from))
+	}
+}
+
+// TestOperationsForServiceIndex asserts the per-service operations index matches
+// a filter over the full Operations map.
+func TestOperationsForServiceIndex(t *testing.T) {
+	g := newTestGraphRAG(t)
+	now := time.Now()
+	s := g.storesForTenant("t").service
+
+	s.UpsertOperation("a", "GET /x", 1, false, now)
+	s.UpsertOperation("a", "GET /y", 1, false, now)
+	s.UpsertOperation("b", "POST /z", 1, false, now)
+	s.UpsertOperation("a", "GET /x", 1, true, now) // repeat: no new index entry
+
+	if ops := s.OperationsForService("a"); len(ops) != 2 {
+		t.Fatalf("OperationsForService(a) = %d, want 2", len(ops))
+	}
+	if ops := s.OperationsForService("b"); len(ops) != 1 {
+		t.Fatalf("OperationsForService(b) = %d, want 1", len(ops))
+	}
+	if ops := s.OperationsForService("missing"); ops != nil {
+		t.Fatalf("OperationsForService(missing) = %v, want nil", ops)
+	}
+}
+
+// TestServiceMapAroundBounded asserts the focused service map only returns the
+// subgraph reachable within depth hops, while the unfocused full map returns all
+// services.
+func TestServiceMapAroundBounded(t *testing.T) {
+	g := newTestGraphRAG(t)
+	now := time.Now()
+	ctx := context.Background()
+	s := g.storesFor(ctx).service
+
+	// Chain a→b→c→d plus an isolated island e→f.
+	for _, sv := range []string{"a", "b", "c", "d", "e", "f"} {
+		s.UpsertService(sv, 1, false, now)
+	}
+	s.UpsertCallEdge("a", "b", 1, false, now)
+	s.UpsertCallEdge("b", "c", 1, false, now)
+	s.UpsertCallEdge("c", "d", 1, false, now)
+	s.UpsertCallEdge("e", "f", 1, false, now)
+
+	// depth 1 from b → {b, a, c}.
+	got := g.ServiceMapAround(ctx, "b", 1)
+	if len(got) != 3 {
+		t.Fatalf("ServiceMapAround(b, 1) = %d entries, want 3", len(got))
+	}
+
+	// depth 2 from a → {a, b, c} (downstream only reaches c at 2 hops).
+	got = g.ServiceMapAround(ctx, "a", 2)
+	if len(got) != 3 {
+		t.Fatalf("ServiceMapAround(a, 2) = %d entries, want 3", len(got))
+	}
+
+	// Unknown seed → nil.
+	if got := g.ServiceMapAround(ctx, "nope", 3); got != nil {
+		t.Fatalf("ServiceMapAround(nope) = %v, want nil", got)
+	}
+
+	// Full map returns every service (island included).
+	if full := g.ServiceMap(ctx, 0); len(full) != 6 {
+		t.Fatalf("ServiceMap = %d entries, want 6", len(full))
+	}
+}
+
+// BenchmarkServiceMap exercises the full topology dump at 200 services / ~800
+// edges to confirm the adjacency index keeps it linear.
+func BenchmarkServiceMap(b *testing.B) {
+	g := newTestGraphRAG(b)
+	now := time.Now()
+	ctx := context.Background()
+	s := g.storesFor(ctx).service
+
+	const n = 200
+	for i := 0; i < n; i++ {
+		s.UpsertService(fmt.Sprintf("svc-%d", i), 1, false, now)
+	}
+	// ~4 outgoing edges per service.
+	for i := 0; i < n; i++ {
+		for k := 1; k <= 4; k++ {
+			s.UpsertCallEdge(fmt.Sprintf("svc-%d", i), fmt.Sprintf("svc-%d", (i+k)%n), 1, false, now)
+		}
+		s.UpsertOperation(fmt.Sprintf("svc-%d", i), "GET /", 1, false, now)
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_ = g.ServiceMap(ctx, 0)
+	}
+}

internal/graphrag/builder_test.go

@@ -28,7 +28,7 @@ func newTestRepo(t *testing.T) *storage.Repository {
 // newTestGraphRAG constructs a GraphRAG usable in tests without a repo or
 // vectordb. The event workers are started so ingestion callbacks process
 // events asynchronously; tests must call Stop() via t.Cleanup.
-func newTestGraphRAG(t *testing.T) *GraphRAG {
+func newTestGraphRAG(t testing.TB) *GraphRAG {
 	t.Helper()
 	g := New(nil, nil, nil, DefaultConfig())
 	// Start only the event workers — the background refresh/snapshot/anomaly

internal/graphrag/queries.go

@@ -383,23 +383,74 @@
 	result := make([]ServiceMapEntry, 0, len(services))
 
 	for _, svc := range services {
-		entry := ServiceMapEntry{
-			Service:  svc,
-			CallsTo:  stores.service.CallEdgesFrom(svc.Name),
-			CalledBy: stores.service.CallEdgesTo(svc.Name),
-		}
+		result = append(result, g.serviceMapEntry(stores, svc))
+	}
+
+	return result
+}
+
+// serviceMapEntry builds the topology entry for one service via the adjacency
+// indexes (O(deg + ops_of_svc), not a full-store scan).
+func (g *GraphRAG) serviceMapEntry(stores *tenantStores, svc *ServiceNode) ServiceMapEntry {
+	return ServiceMapEntry{
+		Service:    svc,
+		CallsTo:    stores.service.CallEdgesFrom(svc.Name),
+		CalledBy:   stores.service.CallEdgesTo(svc.Name),
+		Operations: stores.service.OperationsForService(svc.Name),
+	}
+}
+
+// ServiceMapAround returns the subgraph reachable from seed within depth hops in
+// either direction (downstream via CALLS-from, upstream via CALLS-to). It bounds
+// both the compute and the payload for focused "map around service X" queries at
+// 100–200 services; callers wanting the full map use ServiceMap. depth<=0 falls
+// back to a single hop so a focus query always returns the seed + neighbours.
+func (g *GraphRAG) ServiceMapAround(ctx context.Context, seed string, depth int) []ServiceMapEntry {
+	if depth <= 0 {
+		depth = 1
+	}
+	stores := g.storesFor(ctx)
+	if _, ok := stores.service.GetService(seed); !ok {
+		return nil
+	}
+
+	visited := map[string]bool{seed: true}
+	type queueItem struct {
+		svc   string
+		depth int
+	}
+	queue := []queueItem{{seed, 0}}
+	order := []string{seed}
 
-		// Get operations for this service
-		stores.service.mu.RLock()
-		for _, op := range stores.service.Operations {
-			if op.Service == svc.Name {
-				entry.Operations = append(entry.Operations, op)
+	for len(queue) > 0 {
+		item := queue[0]
+		queue = queue[1:]
+		if item.depth >= depth {
+			continue
+		}
+		neighbours := stores.service.CallEdgesFrom(item.svc)
+		neighbours = append(neighbours, stores.service.CallEdgesTo(item.svc)...)
+		for _, e := range neighbours {
+			next := e.ToID
+			if e.FromID != item.svc {
+				next = e.FromID
 			}
+			if visited[next] {
+				continue
+			}
+			visited[next] = true
+			order = append(order, next)
+			queue = append(queue, queueItem{next, item.depth + 1})
 		}
-		stores.service.mu.RUnlock()
-
-		result = append(result, entry)
 	}
 
+	result := make([]ServiceMapEntry, 0, len(order))
+	for _, name := range order {
+		svc, ok := stores.service.GetService(name)
+		if !ok {
+			continue
+		}
+		result = append(result, g.serviceMapEntry(stores, svc))
+	}
 	return result
 }