Skip to content

Commit d738863

Browse files
aksOpsclaude
andauthored
perf(graphrag,mcp,storage,ui): scale topology to 100-200 services + labelled, searchable map (#124)
* docs(security): sync .bestpractices.json to live passing state (12646) The live OpenSSF record (https://www.bestpractices.dev/projects/12646.json) is now passing (badge_level=passing, 100%). Reconcile the repo golden source to match the authoritative live schema: - Drop 29 vestigial per-criterion <crit>_url keys. The live record has NO per-criterion URL field (only the top-level homepage_url / repo_url / badge_url / project_page_url), confirming the URL must live inline in the justification -- which it now does for all url-required criteria (PR #122). The 4 real top-level URL metadata fields are kept. - Correct _comment to document the inline-URL reality (no separate per-criterion URL field) and drop the false "<key>_url where required" clause. - Simplify report_archive link to /issues to match the live passing value. - Refresh self_assessment_date to 2026-06-18. Kept our richer project-specific justifications rather than overwriting them with bestpractices.dev's auto-generated boilerplate (and a stray OSPS tag); both validate and the badge stays passing. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LDQVJrixs2nJoea67a8pEG * perf(graphrag,mcp,storage,ui): scale topology to 100-200 services + labelled, searchable map Hardens both the SQLite (single-node) and Postgres (scaled) paths for a rolling 7-day window over 100-200 services, and makes the service map usable at that scale. Backend - graphrag: add edgesByFrom/edgesByTo/opsByService adjacency indexes to ServiceStore, maintained at UpsertCallEdge/UpsertOperation/EnsureCallEdge. CallEdgesFrom/To and the new OperationsForService become O(deg) lookups, so ServiceMap drops from O(N*E+N*ops) to O(N+E+ops) and ImpactAnalysis BFS from O(N^2) to O(N+E). ServiceMap at 200 svc / 800 edges benchmarks ~53us. - storage: replace the near-useless bare span.status index with composite (tenant_id, status, start_time); unify the trace/orphan-span purge onto the batched LIMIT loop for ALL drivers so SQLite no longer holds the single writer lock for the whole retention sweep and stalls ingest. - storage: add Postgres BRIN indexes on spans(start_time)/traces(timestamp) for the by-age retention DELETE; raise pool MaxIdleConns 10->25 and add ConnMaxIdleTime (DB_CONN_MAX_IDLE_TIME, default 10m). - config: default IngestPipelinePerTenantCap to ~30% of the resolved queue when unset (explicit =0 still disables) so one noisy tenant cannot fill the queue. MCP - emit compact JSON (Marshal vs MarshalIndent) across all tool responses. - honour the documented get_service_map "service" focus arg via depth-bounded ServiceMapAround; no focus -> full map (backward compatible). - cache trace_graph (immutable completed-trace topology; errors not cached). UI - service map dots now render a haloed, truncated, a11y label whose visibility is zoom-scaled by phyllotaxis criticality band (most-critical label first; emphasized/search-matched always label). - add a deferred client-side service search that highlights matches + neighbours and dims the rest, reusing the existing emphasis machinery, with match count, empty state, and clear. Tests: graphrag/mcp/storage/config/ingest suites green; UI builds + 339 UI tests pass; changed files lint-clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EGHhZg96LhYRPQoG8x4nyP * fix(deps): bump undici 7.25.0 -> 7.28.0 (dev) to clear OSV advisories undici 7.25.0 — a transitive dev dependency via the Vite/test toolchain in ui/package-lock.json — is flagged by two newly-published advisories: GHSA-vmh5-mc38-953g (High, 7.4) and GHSA-pr7r-676h-xcf6 (Medium, 5.9), both fixed in 7.28.0. In-range bump (the parent already allows ^7.25.0), lockfile only. npm audit reports 0 vulnerabilities; tsc + vitest (339 tests) green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LDQVJrixs2nJoea67a8pEG * style(storage): gofmt Span struct tag alignment The composite-index comment added mid-struct in this PR broke gofmt tag alignment on the Span block (golangci-lint gofmt failure). Pure formatting; no behaviour change. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LDQVJrixs2nJoea67a8pEG * fix(ui): highlight matched services' edges on map search, not just nodes rfEdges computed the active path only from selection/impact, so a search accented matched NODES while their connections stayed at resting style — the 'searchable map' was half-done at 100-200 services where edges are the signal. Fold searchMatches into the edge active test (precedence: impact > search > selection, mirroring the node emphasis ladder) and into the memo deps. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LDQVJrixs2nJoea67a8pEG * fix(storage): sanitize DB_DRIVER before logging (CWE-117 / go/log-injection) CodeQL flagged factory.go:170 logging the operator-set DB_DRIVER string (rule go/log-injection, medium) — the merge-blocking unresolved code-scanning thread. Strip CR/LF via a local var before logging so a malformed value can't forge log lines; the counts/duration are non-injectable. Operator config (not a remote-input path), but fix-over-suppress per the security policy, and the local-var sanitizer is a barrier CodeQL recognises, clearing the alert at the source. Drops the now-redundant #nosec G706. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LDQVJrixs2nJoea67a8pEG --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent 56d4e91 commit d738863

15 files changed

Lines changed: 568 additions & 88 deletions

internal/config/config.go

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -185,11 +185,12 @@ type Config struct {
185185
IngestPipelineWorkers int // default 8 worker goroutines
186186
// IngestPipelinePerTenantCap caps in-flight batches per tenant so a noisy
187187
// tenant cannot starve siblings of fresh queue slots when fullness is
188-
// below the soft-backpressure threshold. 0 (default) disables — single-
189-
// tenant deployments need no cap. Operators on multi-tenant deployments
190-
// should set INGEST_PIPELINE_PER_TENANT_CAP to roughly Capacity/N where
191-
// N is the expected number of concurrently-active tenants, with some
192-
// headroom (e.g. 2× the fair-share value) for short bursts.
188+
// below the soft-backpressure threshold. When unset it defaults to ~30% of
189+
// the resolved queue size (see Load) so multi-tenant deployments are
190+
// protected out of the box; an explicit INGEST_PIPELINE_PER_TENANT_CAP=0
191+
// disables the cap for single-tenant deployments. Operators can instead
192+
// pin it to roughly Capacity/N where N is the expected number of
193+
// concurrently-active tenants, with headroom for short bursts.
193194
IngestPipelinePerTenantCap int
194195

195196
// TLS (HTTP + gRPC). When both paths are set, TLS is enabled on both servers.
@@ -377,6 +378,16 @@ func Load(customPath string) (*Config, error) {
377378
AllowSqliteProd: parseTruthy(getEnv("OTELCONTEXT_ALLOW_SQLITE_PROD", "")),
378379
}
379380
applyDriverDefaults(cfg)
381+
382+
// Derive a sane per-tenant ingest cap when the operator did not set one.
383+
// Run AFTER applyDriverDefaults so it tracks the (possibly SQLite-adjusted)
384+
// queue size: ~30% of the queue lets a single tenant burst but stops one
385+
// noisy tenant from monopolising every slot at 100–200 services. An explicit
386+
// INGEST_PIPELINE_PER_TENANT_CAP=0 is respected as "disabled".
387+
if _, set := os.LookupEnv("INGEST_PIPELINE_PER_TENANT_CAP"); !set && cfg.IngestPipelinePerTenantCap == 0 {
388+
cfg.IngestPipelinePerTenantCap = cfg.IngestPipelineQueueSize * 30 / 100
389+
}
390+
380391
return cfg, nil
381392
}
382393

Lines changed: 169 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,169 @@
1+
package graphrag
2+
3+
import (
4+
"context"
5+
"fmt"
6+
"sort"
7+
"testing"
8+
"time"
9+
)
10+
11+
// edgeIDs collects from→to keys so two []*Edge slices can be compared
12+
// order-independently.
13+
func edgeIDs(edges []*Edge) []string {
14+
out := make([]string, 0, len(edges))
15+
for _, e := range edges {
16+
out = append(out, e.FromID+"->"+e.ToID)
17+
}
18+
sort.Strings(out)
19+
return out
20+
}
21+
22+
func fullScanFrom(s *ServiceStore, service string) []*Edge {
23+
var out []*Edge
24+
for _, e := range s.Edges {
25+
if e.Type == EdgeCalls && e.FromID == service {
26+
out = append(out, e)
27+
}
28+
}
29+
return out
30+
}
31+
32+
func fullScanTo(s *ServiceStore, service string) []*Edge {
33+
var out []*Edge
34+
for _, e := range s.Edges {
35+
if e.Type == EdgeCalls && e.ToID == service {
36+
out = append(out, e)
37+
}
38+
}
39+
return out
40+
}
41+
42+
// TestAdjacencyIndexMatchesFullScan asserts the indexed CallEdgesFrom/To return
43+
// exactly the same CALLS edges a full Edges-map scan would, across both the
44+
// aggregating UpsertCallEdge path and the existence-only EnsureCallEdge path.
45+
func TestAdjacencyIndexMatchesFullScan(t *testing.T) {
46+
g := newTestGraphRAG(t)
47+
now := time.Now()
48+
s := g.storesForTenant("t").service
49+
50+
// Aggregating path.
51+
s.UpsertCallEdge("a", "b", 5, false, now)
52+
s.UpsertCallEdge("a", "c", 5, true, now)
53+
s.UpsertCallEdge("b", "c", 5, false, now)
54+
// Existence-only (pre-sample topology observer) path.
55+
s.EnsureService("d", now)
56+
s.EnsureCallEdge("a", "d", now)
57+
s.EnsureCallEdge("d", "c", now)
58+
// Repeat upserts must not duplicate index entries.
59+
s.UpsertCallEdge("a", "b", 5, false, now)
60+
s.EnsureCallEdge("a", "d", now)
61+
62+
for _, svc := range []string{"a", "b", "c", "d"} {
63+
gotFrom := edgeIDs(s.CallEdgesFrom(svc))
64+
wantFrom := edgeIDs(fullScanFrom(s, svc))
65+
if fmt.Sprint(gotFrom) != fmt.Sprint(wantFrom) {
66+
t.Errorf("CallEdgesFrom(%q) = %v, full scan = %v", svc, gotFrom, wantFrom)
67+
}
68+
gotTo := edgeIDs(s.CallEdgesTo(svc))
69+
wantTo := edgeIDs(fullScanTo(s, svc))
70+
if fmt.Sprint(gotTo) != fmt.Sprint(wantTo) {
71+
t.Errorf("CallEdgesTo(%q) = %v, full scan = %v", svc, gotTo, wantTo)
72+
}
73+
}
74+
75+
// a→b should be a single edge despite the repeated upsert.
76+
if from := s.CallEdgesFrom("a"); len(from) != 3 { // a→b, a→c, a→d
77+
t.Fatalf("CallEdgesFrom(a) = %d edges, want 3", len(from))
78+
}
79+
}
80+
81+
// TestOperationsForServiceIndex asserts the per-service operations index matches
82+
// a filter over the full Operations map.
83+
func TestOperationsForServiceIndex(t *testing.T) {
84+
g := newTestGraphRAG(t)
85+
now := time.Now()
86+
s := g.storesForTenant("t").service
87+
88+
s.UpsertOperation("a", "GET /x", 1, false, now)
89+
s.UpsertOperation("a", "GET /y", 1, false, now)
90+
s.UpsertOperation("b", "POST /z", 1, false, now)
91+
s.UpsertOperation("a", "GET /x", 1, true, now) // repeat: no new index entry
92+
93+
if ops := s.OperationsForService("a"); len(ops) != 2 {
94+
t.Fatalf("OperationsForService(a) = %d, want 2", len(ops))
95+
}
96+
if ops := s.OperationsForService("b"); len(ops) != 1 {
97+
t.Fatalf("OperationsForService(b) = %d, want 1", len(ops))
98+
}
99+
if ops := s.OperationsForService("missing"); ops != nil {
100+
t.Fatalf("OperationsForService(missing) = %v, want nil", ops)
101+
}
102+
}
103+
104+
// TestServiceMapAroundBounded asserts the focused service map only returns the
105+
// subgraph reachable within depth hops, while the unfocused full map returns all
106+
// services.
107+
func TestServiceMapAroundBounded(t *testing.T) {
108+
g := newTestGraphRAG(t)
109+
now := time.Now()
110+
ctx := context.Background()
111+
s := g.storesFor(ctx).service
112+
113+
// Chain a→b→c→d plus an isolated island e→f.
114+
for _, sv := range []string{"a", "b", "c", "d", "e", "f"} {
115+
s.UpsertService(sv, 1, false, now)
116+
}
117+
s.UpsertCallEdge("a", "b", 1, false, now)
118+
s.UpsertCallEdge("b", "c", 1, false, now)
119+
s.UpsertCallEdge("c", "d", 1, false, now)
120+
s.UpsertCallEdge("e", "f", 1, false, now)
121+
122+
// depth 1 from b → {b, a, c}.
123+
got := g.ServiceMapAround(ctx, "b", 1)
124+
if len(got) != 3 {
125+
t.Fatalf("ServiceMapAround(b, 1) = %d entries, want 3", len(got))
126+
}
127+
128+
// depth 2 from a → {a, b, c} (downstream only reaches c at 2 hops).
129+
got = g.ServiceMapAround(ctx, "a", 2)
130+
if len(got) != 3 {
131+
t.Fatalf("ServiceMapAround(a, 2) = %d entries, want 3", len(got))
132+
}
133+
134+
// Unknown seed → nil.
135+
if got := g.ServiceMapAround(ctx, "nope", 3); got != nil {
136+
t.Fatalf("ServiceMapAround(nope) = %v, want nil", got)
137+
}
138+
139+
// Full map returns every service (island included).
140+
if full := g.ServiceMap(ctx, 0); len(full) != 6 {
141+
t.Fatalf("ServiceMap = %d entries, want 6", len(full))
142+
}
143+
}
144+
145+
// BenchmarkServiceMap exercises the full topology dump at 200 services / ~800
146+
// edges to confirm the adjacency index keeps it linear.
147+
func BenchmarkServiceMap(b *testing.B) {
148+
g := newTestGraphRAG(b)
149+
now := time.Now()
150+
ctx := context.Background()
151+
s := g.storesFor(ctx).service
152+
153+
const n = 200
154+
for i := 0; i < n; i++ {
155+
s.UpsertService(fmt.Sprintf("svc-%d", i), 1, false, now)
156+
}
157+
// ~4 outgoing edges per service.
158+
for i := 0; i < n; i++ {
159+
for k := 1; k <= 4; k++ {
160+
s.UpsertCallEdge(fmt.Sprintf("svc-%d", i), fmt.Sprintf("svc-%d", (i+k)%n), 1, false, now)
161+
}
162+
s.UpsertOperation(fmt.Sprintf("svc-%d", i), "GET /", 1, false, now)
163+
}
164+
165+
b.ResetTimer()
166+
for i := 0; i < b.N; i++ {
167+
_ = g.ServiceMap(ctx, 0)
168+
}
169+
}

internal/graphrag/builder_test.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ func newTestRepo(t *testing.T) *storage.Repository {
2828
// newTestGraphRAG constructs a GraphRAG usable in tests without a repo or
2929
// vectordb. The event workers are started so ingestion callbacks process
3030
// events asynchronously; tests must call Stop() via t.Cleanup.
31-
func newTestGraphRAG(t *testing.T) *GraphRAG {
31+
func newTestGraphRAG(t testing.TB) *GraphRAG {
3232
t.Helper()
3333
g := New(nil, nil, nil, DefaultConfig())
3434
// Start only the event workers — the background refresh/snapshot/anomaly

internal/graphrag/queries.go

Lines changed: 64 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -383,23 +383,74 @@ func (g *GraphRAG) ServiceMap(ctx context.Context, depth int) []ServiceMapEntry
383383
result := make([]ServiceMapEntry, 0, len(services))
384384

385385
for _, svc := range services {
386-
entry := ServiceMapEntry{
387-
Service: svc,
388-
CallsTo: stores.service.CallEdgesFrom(svc.Name),
389-
CalledBy: stores.service.CallEdgesTo(svc.Name),
390-
}
386+
result = append(result, g.serviceMapEntry(stores, svc))
387+
}
388+
389+
return result
390+
}
391+
392+
// serviceMapEntry builds the topology entry for one service via the adjacency
393+
// indexes (O(deg + ops_of_svc), not a full-store scan).
394+
func (g *GraphRAG) serviceMapEntry(stores *tenantStores, svc *ServiceNode) ServiceMapEntry {
395+
return ServiceMapEntry{
396+
Service: svc,
397+
CallsTo: stores.service.CallEdgesFrom(svc.Name),
398+
CalledBy: stores.service.CallEdgesTo(svc.Name),
399+
Operations: stores.service.OperationsForService(svc.Name),
400+
}
401+
}
402+
403+
// ServiceMapAround returns the subgraph reachable from seed within depth hops in
404+
// either direction (downstream via CALLS-from, upstream via CALLS-to). It bounds
405+
// both the compute and the payload for focused "map around service X" queries at
406+
// 100–200 services; callers wanting the full map use ServiceMap. depth<=0 falls
407+
// back to a single hop so a focus query always returns the seed + neighbours.
408+
func (g *GraphRAG) ServiceMapAround(ctx context.Context, seed string, depth int) []ServiceMapEntry {
409+
if depth <= 0 {
410+
depth = 1
411+
}
412+
stores := g.storesFor(ctx)
413+
if _, ok := stores.service.GetService(seed); !ok {
414+
return nil
415+
}
416+
417+
visited := map[string]bool{seed: true}
418+
type queueItem struct {
419+
svc string
420+
depth int
421+
}
422+
queue := []queueItem{{seed, 0}}
423+
order := []string{seed}
391424

392-
// Get operations for this service
393-
stores.service.mu.RLock()
394-
for _, op := range stores.service.Operations {
395-
if op.Service == svc.Name {
396-
entry.Operations = append(entry.Operations, op)
425+
for len(queue) > 0 {
426+
item := queue[0]
427+
queue = queue[1:]
428+
if item.depth >= depth {
429+
continue
430+
}
431+
neighbours := stores.service.CallEdgesFrom(item.svc)
432+
neighbours = append(neighbours, stores.service.CallEdgesTo(item.svc)...)
433+
for _, e := range neighbours {
434+
next := e.ToID
435+
if e.FromID != item.svc {
436+
next = e.FromID
397437
}
438+
if visited[next] {
439+
continue
440+
}
441+
visited[next] = true
442+
order = append(order, next)
443+
queue = append(queue, queueItem{next, item.depth + 1})
398444
}
399-
stores.service.mu.RUnlock()
400-
401-
result = append(result, entry)
402445
}
403446

447+
result := make([]ServiceMapEntry, 0, len(order))
448+
for _, name := range order {
449+
svc, ok := stores.service.GetService(name)
450+
if !ok {
451+
continue
452+
}
453+
result = append(result, g.serviceMapEntry(stores, svc))
454+
}
404455
return result
405456
}

0 commit comments

Comments
 (0)