fix(graphrag): bound SignalStore.LogClusters (close the last in-memory RAM leak)

CLAUDE.md

@@ -70,7 +70,7 @@ The `internal/graphrag/` package is the core intelligence layer. It replaces the
 |-------|-------|-------|-----|
 | `ServiceStore` | ServiceNode, OperationNode | CALLS, EXPOSES | Permanent |
 | `TraceStore` | TraceNode, SpanNode | CONTAINS, CHILD_OF | Configurable (default 1h) |
-| `SignalStore` | LogClusterNode, MetricNode | EMITTED_BY, MEASURED_BY, LOGGED_DURING | Permanent |
+| `SignalStore` | LogClusterNode, MetricNode | EMITTED_BY, MEASURED_BY, LOGGED_DURING | 24h TTL + per-tenant caps |
 | `AnomalyStore` | AnomalyNode | PRECEDED_BY, TRIGGERED_BY | 24h |
 
 ### Node Types (7)
@@ -219,7 +219,7 @@ Key settings in `internal/config/config.go`:
 - `LOG_FTS_ENABLED` — when truthy (`true`/`yes`/`on`/`1`), provisions the SQLite FTS5 `logs_fts` virtual table + sync triggers at startup; when false, log-search uses a 24h-clamped LIKE fallback. **Defaults to `true` when `DB_DRIVER=sqlite`** (BM25 is dramatically faster than LIKE on the kept `search_logs` MCP tool) and `false` otherwise. Toggle off and reclaim the ~30% disk overhead via `POST /api/admin/drop_fts` (refused while the flag is on). The vectordb-backed semantic-search path was removed on 2026-05-24.
 - `DLQ_MAX_FILES` (1000), `DLQ_MAX_DISK_MB` (500), `DLQ_MAX_RETRIES` (10)
 - `GRAPHRAG_WORKER_COUNT` (16), `GRAPHRAG_EVENT_QUEUE_SIZE` (100000; **10000 on SQLite**) — sized for 100–200 services; raise further if `otelcontext_graphrag_events_dropped_total` climbs
-- `GRAPHRAG_TRACE_TTL` (`1h`; **`30m` on SQLite**), `GRAPHRAG_MAX_SPANS_PER_TENANT` (500000), `GRAPHRAG_TENANT_IDLE_TTL` (`24h`) — in-memory GraphRAG memory bounds. Spans past the per-tenant cap are skipped from the graph only (DB unaffected; metered as `otelcontext_graphrag_events_dropped_total{signal="span_capacity"}`); tenant store slices idle past the TTL are evicted (default tenant immune, self-healing via the 60s rebuild). SignalStore metrics are bounded to 2000/tenant + 24h TTL (constants).
+- `GRAPHRAG_TRACE_TTL` (`1h`; **`30m` on SQLite**), `GRAPHRAG_MAX_SPANS_PER_TENANT` (500000), `GRAPHRAG_TENANT_IDLE_TTL` (`24h`) — in-memory GraphRAG memory bounds. Spans past the per-tenant cap are skipped from the graph only (DB unaffected; metered as `otelcontext_graphrag_events_dropped_total{signal="span_capacity"}`); tenant store slices idle past the TTL are evicted (default tenant immune, self-healing via the 60s rebuild). SignalStore is bounded per-tenant by a 24h TTL + a cap: metrics 2000/tenant, log clusters 10000/tenant (constants; the log-cluster cap was added because clusters key on service×Drain-template-ID and would otherwise grow unbounded as template IDs churn).
 - `PPROF_ADDR` (`127.0.0.1:6060`) — `net/http/pprof` on a dedicated loopback listener (never the public mux); empty disables. Startup also sets a soft `GOMEMLIMIT` (honors the env var, else 75% of the cgroup/host budget via `internal/membudget`).
 - `INGEST_MIN_SEVERITY` (`INFO`), `STORE_MIN_SEVERITY` (**defaults to `"WARN"` for all drivers**; `""` falls back to same-as-ingest) — two-tier log severity gate. The ingest gate runs at the OTLP receiver and **drops the log entirely** below the threshold (no in-memory enrichment either). The store gate runs at the persist boundary inside the async pipeline (`internal/ingest/pipeline.go:process`) and **only skips the DB row write** — the log still flows through `LogCallback` so GraphRAG Drain template mining and span/trace correlation see it. By default (`STORE_MIN_SEVERITY=WARN`, `INGEST_MIN_SEVERITY=INFO`) INFO logs reach GraphRAG/Drain in-memory but are **not** persisted — the DB only grows with WARN+. To also analyse DEBUG in-memory (not just INFO), set `INGEST_MIN_SEVERITY=DEBUG` (raises in-memory event volume). Setting `STORE_MIN_SEVERITY` ≤ `INGEST_MIN_SEVERITY` is a no-op (logged as a warning at startup). Drops surface via `Pipeline.Stats().StoreFiltered`.
 - `INGEST_ASYNC_ENABLED` (true), `INGEST_PIPELINE_QUEUE_SIZE` (50000), `INGEST_PIPELINE_WORKERS` (8), `INGEST_PIPELINE_MAX_BYTES` (536870912 = 512 MB; **128 MB on SQLite**) — async ingest pipeline (`internal/ingest/pipeline.go`). Hybrid backpressure: <90% accept all, 90–100% drop healthy batches (errors/slow always pass), 100% return gRPC `RESOURCE_EXHAUSTED`. The byte cap bounds queue memory regardless of item count — at the cap even priority batches get `RESOURCE_EXHAUSTED`/429 (a 429 is recoverable, an OOM kill is not); watch `otelcontext_ingest_pipeline_queue_bytes` and reason `bytes_full`. Set `INGEST_ASYNC_ENABLED=false` to revert to synchronous DB writes inside `Export()`. Drops surface as `otelcontext_ingest_pipeline_dropped_total{signal,reason}`.

internal/graphrag/refresh.go

@@ -15,6 +15,13 @@ const (
 	// maxMetricsPerTenant caps each tenant's SignalStore metric map; past
 	// the cap the oldest-LastSeen series are evicted first.
 	maxMetricsPerTenant = 2000
+	// maxLogClustersPerTenant caps each tenant's SignalStore LogCluster map.
+	// Clusters are keyed by service×Drain-template-ID, so without a cap the
+	// map outgrows the (shared) Drain template LRU as template IDs churn and
+	// services multiply. The 24h signalRetention TTL is the primary bound;
+	// this cap is the backstop for high-shape-cardinality streams (~1KB/node
+	// => ~10MB/tenant worst case).
+	maxLogClustersPerTenant = 10000
 	// rebuildRowLimit caps how many span rows a single per-tenant rebuild
 	// pass loads. Hitting it is logged — topology lags until the next tick.
 	rebuildRowLimit = 50000
@@ -45,17 +52,17 @@ func (g *GraphRAG) refreshLoop(ctx context.Context) {
 		case <-ticker.C:
 			g.rebuildAllTenantsFromDB(ctx)
 			pruned := 0
-			prunedMetrics := 0
+			prunedSignals := 0
 			signalCutoff := time.Now().Add(-signalRetention)
 			for _, stores := range g.snapshotTenants() {
 				pruned += stores.traces.Prune()
-				prunedMetrics += stores.signals.Prune(signalCutoff, maxMetricsPerTenant)
+				prunedSignals += stores.signals.Prune(signalCutoff, maxMetricsPerTenant, maxLogClustersPerTenant)
 			}
 			if pruned > 0 {
 				slog.Debug("GraphRAG pruned expired traces/spans", "count", pruned)
 			}
-			if prunedMetrics > 0 {
-				slog.Debug("GraphRAG pruned stale/over-cap metrics", "count", prunedMetrics)
+			if prunedSignals > 0 {
+				slog.Debug("GraphRAG pruned stale/over-cap signals (metrics+clusters)", "count", prunedSignals)
 			}
 			g.pruneOldAnomalies()
 			if evicted := g.evictIdleTenants(); evicted > 0 {

internal/graphrag/signal_prune_test.go

@@ -17,7 +17,7 @@ func TestSignalStore_Prune_DropsStaleMetricsKeepsFresh(t *testing.T) {
 	ss.UpsertMetric("cpu", "old-svc", 1.0, stale)
 	ss.UpsertMetric("cpu", "new-svc", 1.0, now)
 
-	if got := ss.Prune(now.Add(-24*time.Hour), 0); got != 1 {
+	if got := ss.Prune(now.Add(-24*time.Hour), 0, 0); got != 1 {
 		t.Fatalf("Prune removed %d metrics, want 1", got)
 	}
 	if _, ok := ss.Metrics["cpu|old-svc"]; ok {
@@ -46,7 +46,7 @@ func TestSignalStore_Prune_CapEvictsOldestFirst(t *testing.T) {
 		ss.UpsertMetric(fmt.Sprintf("m%d", i), "svc", 1.0, now.Add(time.Duration(i)*time.Minute))
 	}
 
-	if got := ss.Prune(now.Add(-24*time.Hour), 2); got != 3 {
+	if got := ss.Prune(now.Add(-24*time.Hour), 2, 0); got != 3 {
 		t.Fatalf("Prune removed %d metrics, want 3", got)
 	}
 	for i := 0; i < 3; i++ {
@@ -65,10 +65,12 @@ func TestSignalStore_Prune_CapEvictsOldestFirst(t *testing.T) {
 	}
 }
 
-// TestSignalStore_Prune_SweepsStaleLogClusterEdgesOnly proves LogClusters
-// themselves are untouched (they are Drain-bounded upstream) while their
-// stale EMITTED_BY / LOGGED_DURING edges are swept.
-func TestSignalStore_Prune_SweepsStaleLogClusterEdgesOnly(t *testing.T) {
+// TestSignalStore_Prune_DropsStaleLogClustersKeepsFresh proves the TTL bound on
+// LogClusters: a cluster idle past the cutoff is removed together with its
+// EMITTED_BY / LOGGED_DURING edges, while a fresh cluster and its edge survive.
+// LogClusters were previously never pruned (only their stale edges were swept),
+// which leaked the map under a sustained high-shape log stream.
+func TestSignalStore_Prune_DropsStaleLogClustersKeepsFresh(t *testing.T) {
 	ss := newSignalStore()
 	now := time.Now()
 	stale := now.Add(-48 * time.Hour)
@@ -77,19 +79,53 @@ func TestSignalStore_Prune_SweepsStaleLogClusterEdgesOnly(t *testing.T) {
 	ss.AddLoggedDuringEdge("c-old", "span-old", stale)
 	ss.UpsertLogCluster("c-new", "tmpl", "ERROR", "svc2", now)
 
-	ss.Prune(now.Add(-24*time.Hour), 0)
+	ss.Prune(now.Add(-24*time.Hour), 0, 0)
 
-	if len(ss.LogClusters) != 2 {
-		t.Fatalf("LogClusters len = %d, want 2 — Prune must not touch clusters", len(ss.LogClusters))
+	if _, ok := ss.LogClusters["c-old"]; ok {
+		t.Errorf("stale LogCluster survived Prune")
+	}
+	if _, ok := ss.LogClusters["c-new"]; !ok {
+		t.Errorf("fresh LogCluster was pruned")
 	}
 	if _, ok := ss.Edges[edgeKey(EdgeEmittedBy, "c-old", "svc")]; ok {
-		t.Errorf("stale EMITTED_BY edge survived")
+		t.Errorf("evicted cluster's EMITTED_BY edge survived")
 	}
 	if _, ok := ss.Edges[edgeKey(EdgeLoggedDuring, "c-old", "span-old")]; ok {
-		t.Errorf("stale LOGGED_DURING edge survived")
+		t.Errorf("evicted cluster's LOGGED_DURING edge survived")
 	}
 	if _, ok := ss.Edges[edgeKey(EdgeEmittedBy, "c-new", "svc2")]; !ok {
-		t.Errorf("fresh EMITTED_BY edge was swept")
+		t.Errorf("fresh cluster's EMITTED_BY edge was swept")
+	}
+}
+
+// TestSignalStore_Prune_LogClusterCapEvictsOldest proves the cap backstop: when
+// the cluster map still exceeds maxLogClusters after the TTL pass, the
+// oldest-LastSeen clusters are evicted first — with their edges — even though
+// all of them are fresh.
+func TestSignalStore_Prune_LogClusterCapEvictsOldest(t *testing.T) {
+	ss := newSignalStore()
+	now := time.Now()
+
+	for i := 0; i < 5; i++ {
+		id := fmt.Sprintf("c%d", i)
+		ss.UpsertLogCluster(id, "tmpl", "ERROR", "svc", now.Add(time.Duration(i)*time.Minute))
+	}
+
+	ss.Prune(now.Add(-24*time.Hour), 0, 2)
+
+	for i := 0; i < 3; i++ {
+		id := fmt.Sprintf("c%d", i)
+		if _, ok := ss.LogClusters[id]; ok {
+			t.Errorf("oldest cluster %s survived cap eviction", id)
+		}
+		if _, ok := ss.Edges[edgeKey(EdgeEmittedBy, id, "svc")]; ok {
+			t.Errorf("cap-evicted cluster %s left a dangling EMITTED_BY edge", id)
+		}
+	}
+	for i := 3; i < 5; i++ {
+		if _, ok := ss.LogClusters[fmt.Sprintf("c%d", i)]; !ok {
+			t.Errorf("newest cluster c%d was evicted — cap must drop oldest first", i)
+		}
 	}
 }
 
@@ -110,7 +146,7 @@ func TestSignalStore_UpsertRefreshesEdgeTimestamps(t *testing.T) {
 	ss.AddLoggedDuringEdge("c1", "span-1", stale)
 	ss.AddLoggedDuringEdge("c1", "span-1", now)
 
-	ss.Prune(now.Add(-24*time.Hour), 0)
+	ss.Prune(now.Add(-24*time.Hour), 0, 0)
 
 	for _, ek := range []string{
 		edgeKey(EdgeMeasuredBy, "cpu|svc", "svc"),

internal/graphrag/store.go

@@ -603,15 +603,16 @@
 	return out
 }
 
-// Prune bounds the SignalStore (modeled on TraceStore.Prune): MetricNodes
-// whose LastSeen predates cutoff are removed; if the map still exceeds
-// maxMetrics (<=0 = uncapped) the oldest-LastSeen overflow is evicted. Each
-// removed metric takes its MEASURED_BY edge with it. Finally, Edges whose
-// UpdatedAt predates cutoff are swept — the upsert paths refresh edge
-// timestamps, so live correlations survive. LogClusters are NOT pruned
-// here: they are bounded upstream by the Drain template LRU; only their
-// stale edges go. Returns the number of metrics removed.
-func (ss *SignalStore) Prune(cutoff time.Time, maxMetrics int) int {
+// Prune bounds the SignalStore (modeled on TraceStore.Prune): MetricNodes and
+// LogClusterNodes whose LastSeen predates cutoff are removed; if either map
+// still exceeds its cap (maxMetrics / maxLogClusters; <=0 = uncapped) the
+// oldest-LastSeen overflow is evicted. Each removed node takes its edges with
+// it — metric MEASURED_BY edges inline, evicted-cluster EMITTED_BY /
+// LOGGED_DURING edges in the final sweep — and any edge whose UpdatedAt
+// predates cutoff is swept too (upsert paths refresh edge timestamps, so live
+// correlations survive). Returns the number of nodes (metrics + clusters)
+// removed.
+func (ss *SignalStore) Prune(cutoff time.Time, maxMetrics, maxLogClusters int) int {
 	ss.mu.Lock()
 	defer ss.mu.Unlock()
 
@@ -640,8 +641,40 @@
 			pruned++
 		}
 	}
+	// LogClusters: TTL + cap, mirroring the metric bound above. This map was
+	// previously unbounded — keyed by service×Drain-template-ID with no cap, it
+	// grew with every distinct (service, template-id) pair ever seen (template
+	// IDs shift as Drain generalizes), leaking under a sustained high-shape log
+	// stream. The TTL ages out that accretion; the cap is the backstop. Evicted
+	// cluster IDs are collected so their EMITTED_BY / LOGGED_DURING edges
+	// (FromID = cluster id) are swept below even when not yet time-stale.
+	evictedClusters := make(map[string]struct{})
+	for id, lc := range ss.LogClusters {
+		if lc.LastSeen.Before(cutoff) {
+			delete(ss.LogClusters, id)
+			evictedClusters[id] = struct{}{}
+			pruned++
+		}
+	}
+	if maxLogClusters > 0 && len(ss.LogClusters) > maxLogClusters {
+		type clusterAge struct {
+			id   string
+			seen time.Time
+		}
+		byAge := make([]clusterAge, 0, len(ss.LogClusters))
+		for id, lc := range ss.LogClusters {
+			byAge = append(byAge, clusterAge{id, lc.LastSeen})
+		}
+		sort.Slice(byAge, func(i, j int) bool { return byAge[i].seen.Before(byAge[j].seen) })
+		for _, e := range byAge[:len(byAge)-maxLogClusters] {
+			delete(ss.LogClusters, e.id)
+			evictedClusters[e.id] = struct{}{}
+			pruned++
+		}
+	}
+
 	for ek, e := range ss.Edges {
-		if e.UpdatedAt.Before(cutoff) {
+		if _, fromEvicted := evictedClusters[e.FromID]; e.UpdatedAt.Before(cutoff) || fromEvicted {
 			delete(ss.Edges, ek)
 		}
 	}