From b54f04e7b744ead70ff3eca9be5dee140935faf7 Mon Sep 17 00:00:00 2001 From: Amit Kumar Date: Thu, 18 Jun 2026 09:54:11 +0000 Subject: [PATCH 1/2] fix(graphrag): observe service topology independent of sampling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cross-service CALLS edges in the service map were only observed when BOTH the caller and callee spans survived the per-service sampler. The sampler runs at the top of TraceServer.Export() and `continue`s past dropped spans before the span model is built or handed to spanCallback/GraphRAG, so at low sample rates the joint survival of a parent+child pair is vanishingly small and the in-memory topology is starved — service nodes render with almost no edges (no visible flow direction). The 60s DB rebuild cannot recover it because the DB also only ever received sampled spans. Add a bounded, pre-sample topology observer in the graphrag package. The OTLP trace receiver now calls GraphRAG.ObserveSpanTopology for EVERY received span BEFORE the sampler's keep/drop decision, on both the async pipeline path and the synchronous fallback. The observer maintains a per-tenant bounded LRU (map[spanID]serviceName, cap 100k, container/list LRU mirroring drain.go) and, for each span whose parent is a known span in a different service, guarantees the parentService->childService CALLS edge EXISTS — deduped per (src->tgt) pair so work and memory are bounded by the number of distinct service-pairs. Edge existence is guaranteed via new existence-only ServiceStore methods EnsureService and EnsureCallEdge, which create the node/edge with zeroed aggregates if absent and are no-ops otherwise. Edge aggregates (CallCount/latency/error-rate) remain exclusively owned by the sampled path (processSpan -> UpsertCallEdge) and the 60s DB rebuild, so the observer never pollutes them. Error/slow always-on sampling, DB persistence (sampled), and per-tenant store isolation are unchanged. The observer is wired into TraceServer via a function-typed hook (SetTopologyObserver) to keep internal/ingest decoupled from internal/graphrag, matching the existing spanCallback injection pattern; main.go wires it to graphRAG.ObserveSpanTopology. Tests (TDD): - TestObserveSpanTopology_EdgeSurvivesSampling: checkout->payments edge present when no sampled span ever formed it. - TestObserveSpanTopology_PreservesSampledAggregates: observing a pair with an existing sampled edge does not inflate its aggregates. - TestObserveSpanTopology_BoundedLRUEvicts: per-tenant spanID->service LRU evicts oldest past cap. - TestObserveSpanTopology_DedupBoundsEdges: repeated pair observation keeps one edge. - TestObserveSpanTopology_TenantIsolation: edges/parent lookups do not leak across tenants. - TestObserveSpanTopology_{UnknownParentNoEdge,SameServiceNoEdge}: negative cases. - ingest TestTopologyObserver_FiresPreSample_{Async,Sync}Path: the hook fires for both spans before a ~99%-drop sampler on both ingest paths. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01LDQVJrixs2nJoea67a8pEG --- internal/graphrag/store.go | 57 +++++ internal/graphrag/topology_observer.go | 201 +++++++++++++++ internal/graphrag/topology_observer_test.go | 240 ++++++++++++++++++ internal/ingest/otlp.go | 37 ++- .../ingest/topology_observer_wiring_test.go | 162 ++++++++++++ main.go | 6 + 6 files changed, 699 insertions(+), 4 deletions(-) create mode 100644 internal/graphrag/topology_observer.go create mode 100644 internal/graphrag/topology_observer_test.go create mode 100644 internal/ingest/topology_observer_wiring_test.go diff --git a/internal/graphrag/store.go b/internal/graphrag/store.go index 9fb5856..a229a7c 100644 --- a/internal/graphrag/store.go +++ b/internal/graphrag/store.go @@ -19,6 +19,12 @@ type tenantStores struct { signals *SignalStore anomalies *AnomalyStore + // topology records cross-service CALLS edges from EVERY received span, + // pre-sample (see topology_observer.go). Bounded and tenant-isolated by + // living on this slice; guarantees the service map shows flow direction + // even when sampling starves the sampled edge path. + topology *topologyObserver + // lastAccess is the unix-nano time of the most recent ingest event or // query routed through storesForTenant. Drives idle-tenant eviction in // refreshLoop; background maintenance uses tenantStoresNoTouch so a 60s @@ -44,6 +50,7 @@ func newTenantStores(traceTTL time.Duration, maxSpans int) *tenantStores { traces: newTraceStore(traceTTL, maxSpans), signals: newSignalStore(), anomalies: newAnomalyStore(), + topology: newTopologyObserver(), } // Creation counts as access — a slice re-created by the DB rebuild gets // a full idle window rather than being evicted on the next tick. @@ -222,6 +229,56 @@ func (s *ServiceStore) UpsertCallEdge(source, target string, durationMs float64, e.UpdatedAt = ts } +// EnsureService guarantees a ServiceNode for name EXISTS without touching its +// aggregates. Absent → created with zeroed call/error stats (FirstSeen/LastSeen +// = ts); present → no-op. The pre-sample topology observer uses this so the +// service map has a row to hang flow-direction edges on even when sampling +// dropped every span for the service; the sampled path (UpsertService) and the +// 60s DB rebuild remain the sole source of call/error/latency aggregates. +// Returns true if a new node was created. +func (s *ServiceStore) EnsureService(name string, ts time.Time) bool { + if name == "" { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + + if _, ok := s.Services[name]; ok { + return false + } + s.Services[name] = &ServiceNode{ + ID: name, + Name: name, + FirstSeen: ts, + LastSeen: ts, + } + return true +} + +// EnsureCallEdge guarantees a CALLS edge source→target EXISTS without touching +// its aggregates. If the edge is absent it is created with zeroed +// CallCount/latency/error stats (UpdatedAt = ts); if it already exists this is a +// no-op. The pre-sample topology observer uses this so the service map shows +// flow direction even when sampling dropped every span that would have formed +// the edge — while the sampled path (UpsertCallEdge) remains the sole source of +// CallCount/latency/error-rate aggregates. Returns true if a new edge was created. +func (s *ServiceStore) EnsureCallEdge(source, target string, ts time.Time) bool { + ek := edgeKey(EdgeCalls, source, target) + s.mu.Lock() + defer s.mu.Unlock() + + if _, ok := s.Edges[ek]; ok { + return false + } + s.Edges[ek] = &Edge{ + Type: EdgeCalls, + FromID: source, + ToID: target, + UpdatedAt: ts, + } + return true +} + func (s *ServiceStore) GetService(name string) (*ServiceNode, bool) { s.mu.RLock() defer s.mu.RUnlock() diff --git a/internal/graphrag/topology_observer.go b/internal/graphrag/topology_observer.go new file mode 100644 index 0000000..a7a225e --- /dev/null +++ b/internal/graphrag/topology_observer.go @@ -0,0 +1,201 @@ +package graphrag + +import ( + "container/list" + "sync" + "time" +) + +// topologyObserverSpanCap bounds the per-tenant spanID→service LRU. Cross-service +// edges are inferred by joining a child span's ParentSpanID to its parent's +// service, so we must remember recently-seen span→service mappings long enough +// for the child to arrive. Spans within a trace arrive close together, so a +// window of ~100k recent spans per tenant comfortably covers in-flight traces +// while bounding memory: each entry is two short strings (~0.1 KB incl. map and +// list overhead), so the cap is ≲ a few tens of MB per tenant worst case. +// +// This is the LRU pattern already used by the Drain miner (drain.go): a +// container/list for O(1) recency + a map for O(1) lookup, evicting from the +// back when over capacity. +const topologyObserverSpanCap = 100_000 + +// topologyObserver records cross-service CALLS edges INDEPENDENT of the trace +// sampler's keep/drop decision. The OTLP receiver invokes ObserveSpanTopology +// for every received span BEFORE the sampler runs; without this, an edge is +// only ever formed when BOTH the parent and child spans survive sampling — at +// low sample rates the joint survival of a pair is vanishingly small, so the +// in-memory service map shows nodes but almost no edges (no flow direction). +// +// It is deliberately lightweight and strictly bounded: +// - a per-tenant LRU map[spanID]serviceName (cap topologyObserverSpanCap) +// so memory is bounded by recent span volume, not total spans seen; +// - a per-tenant set of already-emitted (source→target) pairs so repeated +// observations of the same edge do no work and the set is bounded by the +// number of distinct service-pairs (~services²), not span count. +// +// Edge existence is guaranteed via ServiceStore.EnsureCallEdge, which creates +// the edge with zeroed aggregates if absent and is a no-op otherwise. Edge +// aggregates (CallCount/latency/error-rate) remain exclusively owned by the +// sampled path (processSpan → UpsertCallEdge) and the 60s DB rebuild, so this +// observer never pollutes them. +// +// Each topologyObserver belongs to exactly one tenantStores slice, so its data +// is tenant-isolated by construction and inherits the slice's idle eviction. +type topologyObserver struct { + mu sync.Mutex + + // spanService is a bounded LRU of spanID → serviceName for recently seen + // spans. order tracks recency (front = most recent); elems maps spanID to + // its list element for O(1) move/remove. The list element's Value is the + // spanID string. + spanService map[string]string + order *list.List + elems map[string]*list.Element + + // emittedPairs deduplicates (source→target) so observing a hot edge + // thousands of times costs one EnsureCallEdge and one map write total. + emittedPairs map[string]struct{} + + // emittedServices deduplicates service-node creation so the ServiceStore + // write lock is taken once per new service, not once per span. Bounded by + // the number of distinct services in the tenant. + emittedServices map[string]struct{} + + cap int +} + +func newTopologyObserver() *topologyObserver { + return &topologyObserver{ + spanService: make(map[string]string), + order: list.New(), + elems: make(map[string]*list.Element), + emittedPairs: make(map[string]struct{}), + emittedServices: make(map[string]struct{}), + cap: topologyObserverSpanCap, + } +} + +// observe records span→service and, when the span's parent is a known span in a +// DIFFERENT service, ensures the parentService→service CALLS edge exists in the +// supplied ServiceStore. svc must be non-empty (the caller guarantees this). +func (o *topologyObserver) observe(svcStore *ServiceStore, spanID, parentSpanID, service string) { + if spanID == "" || service == "" { + return + } + + o.mu.Lock() + // Record/refresh this span's service mapping (LRU touch). + o.put(spanID, service) + + // Decide what to materialize while holding the lock so all map reads/writes + // are race-free; apply to the ServiceStore after releasing it. We dedup both + // service-node and edge creation so steady-state observation does no work. + ensureChild := o.markServiceOnce(service) + var ensureSource bool + var emitSource string + if parentSpanID != "" { + if parentService, ok := o.spanService[parentSpanID]; ok && parentService != service { + o.touch(parentSpanID) // parent is still active in this trace + pairKey := parentService + "\x00" + service + if _, done := o.emittedPairs[pairKey]; !done { + o.emittedPairs[pairKey] = struct{}{} + emitSource = parentService + ensureSource = o.markServiceOnce(parentService) + } + } + } + o.mu.Unlock() + + // Existence-only writes: create nodes/edge if absent, no-op otherwise, never + // touching sampled aggregates. + now := time.Now() + if ensureChild { + svcStore.EnsureService(service, now) + } + if ensureSource { + svcStore.EnsureService(emitSource, now) + } + if emitSource != "" { + svcStore.EnsureCallEdge(emitSource, service, now) + } +} + +// markServiceOnce returns true the first time it sees a service name (recording +// it), false thereafter. Caller holds o.mu. Lets the caller take the ServiceStore +// write lock only once per distinct service rather than once per span. +func (o *topologyObserver) markServiceOnce(service string) bool { + if _, ok := o.emittedServices[service]; ok { + return false + } + o.emittedServices[service] = struct{}{} + return true +} + +// put inserts or refreshes spanID→service and enforces the LRU cap. Caller holds o.mu. +func (o *topologyObserver) put(spanID, service string) { + if el, ok := o.elems[spanID]; ok { + o.spanService[spanID] = service + o.order.MoveToFront(el) + return + } + o.spanService[spanID] = service + o.elems[spanID] = o.order.PushFront(spanID) + o.evictIfNeeded() +} + +// touch marks spanID as most-recently used without changing its mapping. Caller holds o.mu. +func (o *topologyObserver) touch(spanID string) { + if el, ok := o.elems[spanID]; ok { + o.order.MoveToFront(el) + } +} + +// evictIfNeeded drops least-recently-used span mappings past the cap. The +// emittedPairs set is intentionally NOT pruned here — it is independently +// bounded by the number of distinct service-pairs and an evicted span does not +// invalidate an already-recorded edge. Caller holds o.mu. +func (o *topologyObserver) evictIfNeeded() { + for o.cap > 0 && len(o.spanService) > o.cap { + back := o.order.Back() + if back == nil { + return + } + victim, _ := back.Value.(string) + o.order.Remove(back) + delete(o.elems, victim) + delete(o.spanService, victim) + } +} + +// len reports the current LRU size. Exported via GraphRAG.topologyObserverLen +// for tests. +func (o *topologyObserver) len() int { + o.mu.Lock() + defer o.mu.Unlock() + return len(o.spanService) +} + +// ObserveSpanTopology records one received span's cross-service call topology +// INDEPENDENT of the sampler. The OTLP trace receiver calls this for every span +// BEFORE the sampler's keep/drop decision, so the service map retains flow +// direction even when sampling drops the spans that would have formed the edge. +// +// tenant is the resolved tenant (empty collapses to DefaultTenantID, matching +// the rest of GraphRAG). traceID is accepted for symmetry with the event path +// and possible future per-trace scoping; the current implementation keys solely +// on span IDs, which are globally unique within a tenant. Safe to call from many +// goroutines. +func (g *GraphRAG) ObserveSpanTopology(tenant, traceID, spanID, parentSpanID, service string) { + if service == "" || spanID == "" { + return + } + stores := g.storesForTenant(tenant) + stores.lastEventAt.Store(time.Now().UnixNano()) + stores.topology.observe(stores.service, spanID, parentSpanID, service) +} + +// topologyObserverLen returns the per-tenant topology observer's current LRU +// size. Test-only accessor. +func (g *GraphRAG) topologyObserverLen(tenant string) int { + return g.storesForTenant(tenant).topology.len() +} diff --git a/internal/graphrag/topology_observer_test.go b/internal/graphrag/topology_observer_test.go new file mode 100644 index 0000000..bac27a4 --- /dev/null +++ b/internal/graphrag/topology_observer_test.go @@ -0,0 +1,240 @@ +package graphrag + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/RandomCodeSpace/otelcontext/internal/storage" +) + +// TestObserveSpanTopology_EdgeSurvivesSampling is the core regression for the +// service-map edge-starvation bug: under aggressive sampling almost no +// parent+child span PAIR survives, so the sampled path (OnSpanIngested → +// processSpan → UpsertCallEdge) never forms the cross-service CALLS edge. +// +// The pre-sample topology observer must guarantee the checkout→payments edge +// EXISTS even though no sampled span ever did. Here we simulate the worst case +// directly: NOT A SINGLE span reaches OnSpanIngested (everything was dropped by +// the sampler upstream), yet the observer saw both spans before the drop. +func TestObserveSpanTopology_EdgeSurvivesSampling(t *testing.T) { + g := newTestGraphRAG(t) + + tenant := storage.DefaultTenantID + // Parent span "checkout" (root: empty parent), child span "payments" whose + // parent is the checkout span. Observed pre-sample, in arrival order. + g.ObserveSpanTopology(tenant, "trace-1", "span-checkout", "", "checkout") + g.ObserveSpanTopology(tenant, "trace-1", "span-payments", "span-checkout", "payments") + + ctx := storage.WithTenantContext(context.Background(), tenant) + + // The observer upserts synchronously (no event channel), so the edge is + // present immediately — no polling needed. + if !hasCallEdge(g.ServiceMap(ctx, 0), "checkout", "payments") { + t.Fatalf("checkout→payments CALLS edge absent; observer failed to record topology independent of sampling") + } +} + +// TestObserveSpanTopology_UnknownParentNoEdge confirms that a child whose +// parent span has not been observed yet produces no edge (we cannot attribute +// the call to a source service). It must not panic or fabricate an edge. +func TestObserveSpanTopology_UnknownParentNoEdge(t *testing.T) { + g := newTestGraphRAG(t) + tenant := storage.DefaultTenantID + + g.ObserveSpanTopology(tenant, "trace-1", "span-child", "span-missing-parent", "payments") + + ctx := storage.WithTenantContext(context.Background(), tenant) + for _, e := range allCallEdges(g.ServiceMap(ctx, 0)) { + t.Fatalf("unexpected edge with unknown parent: %s→%s", e.FromID, e.ToID) + } +} + +// TestObserveSpanTopology_SameServiceNoEdge confirms a parent/child pair in the +// SAME service does not create a self-edge — CALLS is cross-service only. +func TestObserveSpanTopology_SameServiceNoEdge(t *testing.T) { + g := newTestGraphRAG(t) + tenant := storage.DefaultTenantID + + g.ObserveSpanTopology(tenant, "trace-1", "span-a", "", "checkout") + g.ObserveSpanTopology(tenant, "trace-1", "span-b", "span-a", "checkout") + + ctx := storage.WithTenantContext(context.Background(), tenant) + for _, e := range allCallEdges(g.ServiceMap(ctx, 0)) { + t.Fatalf("unexpected self-edge: %s→%s", e.FromID, e.ToID) + } +} + +// TestObserveSpanTopology_PreservesSampledAggregates is the no-regression guard +// for edge statistics. The sampled path computes CallCount/latency/error-rate. +// The observer only guarantees existence: when a sampled span already formed the +// edge, observing the same pair again must NOT inflate the sampled aggregates. +func TestObserveSpanTopology_PreservesSampledAggregates(t *testing.T) { + g := newTestGraphRAG(t) + tenant := storage.DefaultTenantID + now := time.Now() + + // Deterministically establish a checkout→payments edge WITH real aggregates, + // exactly as the sampled path does (processSpan → UpsertCallEdge). Done + // directly on the store so this test does not depend on async event-worker + // ordering (the sampled path needs the parent span resident before the child + // is processed; that ordering is not guaranteed and is orthogonal to the + // invariant under test). + stores := g.storesForTenant(tenant) + stores.service.UpsertService("checkout", 2.0, false, now) + stores.service.UpsertService("payments", 2.0, false, now) + const sampledCalls = 5 + for i := 0; i < sampledCalls; i++ { + stores.service.UpsertCallEdge("checkout", "payments", 2.0, false, now) + } + + ctx := storage.WithTenantContext(context.Background(), tenant) + edge := getCallEdge(g.ServiceMap(ctx, 0), "checkout", "payments") + if edge == nil { + t.Fatalf("setup: sampled-path edge not present") + } + wantCount, wantTotalMs, wantWeight := edge.CallCount, edge.TotalMs, edge.Weight + if wantCount != sampledCalls { + t.Fatalf("setup: CallCount = %d, want %d", wantCount, sampledCalls) + } + + // Now the observer sees the SAME pair many times (as it would pre-sample for + // every received span). Existence-only upsert must leave aggregates intact. + for i := 0; i < 100; i++ { + g.ObserveSpanTopology(tenant, "trace-1", "span-checkout", "", "checkout") + g.ObserveSpanTopology(tenant, "trace-1", "span-payments", "span-checkout", "payments") + } + + edge = getCallEdge(g.ServiceMap(ctx, 0), "checkout", "payments") + if edge.CallCount != wantCount || edge.TotalMs != wantTotalMs || edge.Weight != wantWeight { + t.Fatalf("observer polluted sampled edge aggregates: got {CallCount:%d TotalMs:%g Weight:%g}, want {CallCount:%d TotalMs:%g Weight:%g} (aggregates must come from sampled spans only)", + edge.CallCount, edge.TotalMs, edge.Weight, wantCount, wantTotalMs, wantWeight) + } +} + +// TestObserveSpanTopology_BoundedLRUEvicts confirms the per-tenant +// spanID→service map evicts the oldest entries past its cap, so a long-running +// process cannot grow it without bound. After eviction the evicted parent can no +// longer be resolved, so a late child referencing it forms no edge. +func TestObserveSpanTopology_BoundedLRUEvicts(t *testing.T) { + g := newTestGraphRAG(t) + tenant := storage.DefaultTenantID + cap := topologyObserverSpanCap + + // Observe one "anchor" parent span in service "anchor-svc". + g.ObserveSpanTopology(tenant, "trace-anchor", "anchor-span", "", "anchor-svc") + + // Now observe `cap` further root spans (no parent ⇒ no edges) to push the + // anchor past the LRU window. Each distinct spanID consumes one slot. + for i := 0; i < cap; i++ { + g.ObserveSpanTopology(tenant, "trace-fill", fmt.Sprintf("fill-%d", i), "", "filler-svc") + } + + // A child referencing the now-evicted anchor parent must NOT form an edge, + // because the parent's service is no longer resolvable. + g.ObserveSpanTopology(tenant, "trace-late", "late-child", "anchor-span", "late-svc") + + ctx := storage.WithTenantContext(context.Background(), tenant) + if hasCallEdge(g.ServiceMap(ctx, 0), "anchor-svc", "late-svc") { + t.Fatalf("anchor parent was not evicted: edge formed from an entry that should have aged out of the bounded LRU") + } + + // Sanity: the LRU map size is bounded at the cap. + if got := g.topologyObserverLen(tenant); got > cap { + t.Fatalf("LRU size %d exceeds cap %d", got, cap) + } +} + +// TestObserveSpanTopology_DedupBoundsEdges confirms that observing the same +// service-pair many times keeps the edge set bounded to one edge per pair. +func TestObserveSpanTopology_DedupBoundsEdges(t *testing.T) { + g := newTestGraphRAG(t) + tenant := storage.DefaultTenantID + + // Distinct span IDs but the SAME service pair (checkout→payments) repeated. + for i := 0; i < 500; i++ { + parent := fmt.Sprintf("p-%d", i) + child := fmt.Sprintf("c-%d", i) + g.ObserveSpanTopology(tenant, "trace-x", parent, "", "checkout") + g.ObserveSpanTopology(tenant, "trace-x", child, parent, "payments") + } + + ctx := storage.WithTenantContext(context.Background(), tenant) + edges := allCallEdges(g.ServiceMap(ctx, 0)) + if len(edges) != 1 { + t.Fatalf("expected exactly 1 deduped edge for the single service-pair, got %d: %+v", len(edges), edges) + } + if edges[0].FromID != "checkout" || edges[0].ToID != "payments" { + t.Fatalf("edge = %s→%s, want checkout→payments", edges[0].FromID, edges[0].ToID) + } +} + +// TestObserveSpanTopology_TenantIsolation confirms edges observed under one +// tenant never leak into another tenant's ServiceMap. +func TestObserveSpanTopology_TenantIsolation(t *testing.T) { + g := newTestGraphRAG(t) + + // Tenant A: orders→inventory. + g.ObserveSpanTopology("tenant-a", "t-a", "a-parent", "", "orders") + g.ObserveSpanTopology("tenant-a", "t-a", "a-child", "a-parent", "inventory") + // Tenant B: gateway→auth. + g.ObserveSpanTopology("tenant-b", "t-b", "b-parent", "", "gateway") + g.ObserveSpanTopology("tenant-b", "t-b", "b-child", "b-parent", "auth") + + ctxA := storage.WithTenantContext(context.Background(), "tenant-a") + ctxB := storage.WithTenantContext(context.Background(), "tenant-b") + + if !hasCallEdge(g.ServiceMap(ctxA, 0), "orders", "inventory") { + t.Fatalf("tenant-a missing its own orders→inventory edge") + } + if !hasCallEdge(g.ServiceMap(ctxB, 0), "gateway", "auth") { + t.Fatalf("tenant-b missing its own gateway→auth edge") + } + // Cross-tenant leakage in BOTH directions. + if hasCallEdge(g.ServiceMap(ctxA, 0), "gateway", "auth") { + t.Fatalf("tenant-a leaked tenant-b's gateway→auth edge") + } + if hasCallEdge(g.ServiceMap(ctxB, 0), "orders", "inventory") { + t.Fatalf("tenant-b leaked tenant-a's orders→inventory edge") + } + + // A parent span ID observed only under tenant A must not resolve under + // tenant B (the LRU is per-tenant): a B child reusing "a-parent" forms no edge. + g.ObserveSpanTopology("tenant-b", "t-b", "b-child-2", "a-parent", "billing") + if hasCallEdge(g.ServiceMap(ctxB, 0), "orders", "billing") { + t.Fatalf("tenant-b resolved a parent span that belongs to tenant-a's LRU") + } +} + +// --- test helpers --- + +func allCallEdges(entries []ServiceMapEntry) []*Edge { + seen := make(map[string]*Edge) + for _, e := range entries { + for _, edge := range e.CallsTo { + seen[edge.FromID+"|"+edge.ToID] = edge + } + for _, edge := range e.CalledBy { + seen[edge.FromID+"|"+edge.ToID] = edge + } + } + out := make([]*Edge, 0, len(seen)) + for _, edge := range seen { + out = append(out, edge) + } + return out +} + +func getCallEdge(entries []ServiceMapEntry, from, to string) *Edge { + for _, edge := range allCallEdges(entries) { + if edge.FromID == from && edge.ToID == to { + return edge + } + } + return nil +} + +func hasCallEdge(entries []ServiceMapEntry, from, to string) bool { + return getCallEdge(entries, from, to) != nil +} diff --git a/internal/ingest/otlp.go b/internal/ingest/otlp.go index d78d054..0c784fe 100644 --- a/internal/ingest/otlp.go +++ b/internal/ingest/otlp.go @@ -102,10 +102,15 @@ func tenantFromResource(attrs []*commonpb.KeyValue) string { } type TraceServer struct { - repo *storage.Repository - metrics *telemetry.Metrics - logCallback func(storage.Log) - spanCallback func(storage.Span) // called for each span after persistence + repo *storage.Repository + metrics *telemetry.Metrics + logCallback func(storage.Log) + spanCallback func(storage.Span) // called for each span after persistence + // topologyObserver, when set, is invoked for EVERY received span BEFORE the + // sampler's keep/drop decision so cross-service call topology survives + // sampling (see graphrag.GraphRAG.ObserveSpanTopology). Args: + // tenant, traceID, spanID, parentSpanID, service. nil = disabled. + topologyObserver func(tenant, traceID, spanID, parentSpanID, service string) minSeverity int allowedServices map[string]bool excludedServices map[string]bool @@ -170,6 +175,14 @@ func (s *TraceServer) SetSampler(sm *Sampler) { s.sampler = sm } +// SetTopologyObserver wires a pre-sample hook invoked for every received span +// before the sampler runs, so the in-memory service map keeps cross-service +// flow direction even when sampling drops the spans that would have formed the +// edge. Pass nil to disable. Wired in main.go to graphrag.ObserveSpanTopology. +func (s *TraceServer) SetTopologyObserver(cb func(tenant, traceID, spanID, parentSpanID, service string)) { + s.topologyObserver = cb +} + // SetPipeline enables the async ingest pipeline. When set, Export() // returns to the caller as soon as the parsed batch is enqueued (or // rejected), and persistence runs on the pipeline's worker pool. Pass @@ -336,6 +349,22 @@ func (s *TraceServer) Export(ctx context.Context, req *coltracepb.ExportTraceSer if span.Status != nil { statusStr = span.Status.Code.String() } + + // Observe cross-service call topology for EVERY span BEFORE + // the sampler can drop it. The sampled path below still owns + // edge aggregates; this only guarantees the edge exists so the + // service map keeps flow direction at low sample rates. Cheap + // and strictly bounded (per-tenant LRU + per-pair dedup). + if s.topologyObserver != nil { + s.topologyObserver( + tenantID, + fmt.Sprintf("%x", span.TraceId), + fmt.Sprintf("%x", span.SpanId), + fmt.Sprintf("%x", span.ParentSpanId), + serviceName, + ) + } + if s.sampler != nil { isError := statusStr == "STATUS_CODE_ERROR" durationMs := float64(duration) / 1000.0 diff --git a/internal/ingest/topology_observer_wiring_test.go b/internal/ingest/topology_observer_wiring_test.go new file mode 100644 index 0000000..9a1f622 --- /dev/null +++ b/internal/ingest/topology_observer_wiring_test.go @@ -0,0 +1,162 @@ +package ingest + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/RandomCodeSpace/otelcontext/internal/config" + "github.com/RandomCodeSpace/otelcontext/internal/storage" + + coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1" + commonpb "go.opentelemetry.io/proto/otlp/common/v1" + resourcepb "go.opentelemetry.io/proto/otlp/resource/v1" + tracepb "go.opentelemetry.io/proto/otlp/trace/v1" +) + +// observedPair is a service→service topology observation captured by the test +// hook. We record source/target service so we can assert the cross-service +// CALLS edge was observed regardless of sampler keep/drop decisions. +type observedSpan struct { + tenant, traceID, spanID, parentSpanID, service string +} + +// twoServiceTraceReq builds one trace with a "checkout" root span and a +// "payments" child span (child.ParentSpanId == root.SpanId), across two +// resource_spans so each carries its own service.name resource attribute. +func twoServiceTraceReq() *coltracepb.ExportTraceServiceRequest { + res := func(svc string) *resourcepb.Resource { + return &resourcepb.Resource{Attributes: []*commonpb.KeyValue{{ + Key: "service.name", + Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: svc}}, + }}} + } + now := uint64(time.Now().UnixNano()) + span := func(name string, spanID, parentID []byte) *tracepb.Span { + return &tracepb.Span{ + TraceId: []byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10}, + SpanId: spanID, + ParentSpanId: parentID, + Name: name, + StartTimeUnixNano: now, + EndTimeUnixNano: now + uint64(time.Millisecond), + Status: &tracepb.Status{Code: tracepb.Status_STATUS_CODE_OK}, + } + } + checkoutSpanID := []byte{0xaa, 0xaa, 0xaa, 0xaa, 0xaa, 0xaa, 0xaa, 0xaa} + paymentsSpanID := []byte{0xbb, 0xbb, 0xbb, 0xbb, 0xbb, 0xbb, 0xbb, 0xbb} + return &coltracepb.ExportTraceServiceRequest{ + ResourceSpans: []*tracepb.ResourceSpans{ + { + Resource: res("checkout"), + ScopeSpans: []*tracepb.ScopeSpans{{Spans: []*tracepb.Span{span("/checkout", checkoutSpanID, nil)}}}, + }, + { + Resource: res("payments"), + ScopeSpans: []*tracepb.ScopeSpans{{Spans: []*tracepb.Span{span("/charge", paymentsSpanID, checkoutSpanID)}}}, + }, + }, + } +} + +// newTopologyWiringServer builds a TraceServer with a near-zero sampling rate +// and a recording topology observer. asyncPipeline selects whether the async +// ingest pipeline is wired (true) or the synchronous fallback path is used +// (false) — the observer MUST fire on both before the sampler drop. +func newTopologyWiringServer(t *testing.T, asyncPipeline bool) (*TraceServer, *recorder) { + t.Helper() + + db, err := storage.NewDatabase("sqlite", ":memory:") + if err != nil { + t.Fatalf("NewDatabase: %v", err) + } + if err := storage.AutoMigrateModels(db, "sqlite"); err != nil { + t.Fatalf("AutoMigrateModels: %v", err) + } + repo := storage.NewRepositoryFromDB(db, "sqlite") + t.Cleanup(func() { _ = repo.Close() }) + + cfg := &config.Config{IngestMinSeverity: "DEBUG"} + srv := NewTraceServer(repo, nil, cfg) + + // Aggressively drop healthy spans. rate≈0 means only the per-service burst + // (first span) survives; the joint survival of the parent+child PAIR across + // two services is effectively nil — exactly the starvation scenario. + srv.SetSampler(NewSampler(0.0001, true, 1_000_000)) + + rec := &recorder{} + srv.SetTopologyObserver(rec.observe) + + if asyncPipeline { + p := NewPipeline(repo, nil, PipelineConfig{Capacity: 100, Workers: 1, SoftThreshold: 0.9}) + p.Start(context.Background()) + t.Cleanup(p.Stop) + srv.SetPipeline(p) + } + return srv, rec +} + +func TestTopologyObserver_FiresPreSample_AsyncPath(t *testing.T) { + srv, rec := newTopologyWiringServer(t, true) + + if _, err := srv.Export(context.Background(), twoServiceTraceReq()); err != nil { + t.Fatalf("Export: %v", err) + } + + rec.assertObservedPair(t, "checkout", "payments") +} + +func TestTopologyObserver_FiresPreSample_SyncPath(t *testing.T) { + srv, rec := newTopologyWiringServer(t, false) + + if _, err := srv.Export(context.Background(), twoServiceTraceReq()); err != nil { + t.Fatalf("Export: %v", err) + } + + rec.assertObservedPair(t, "checkout", "payments") +} + +// recorder captures topology observations from the hook under test. +type recorder struct { + mu sync.Mutex + spans []observedSpan +} + +func (r *recorder) observe(tenant, traceID, spanID, parentSpanID, service string) { + r.mu.Lock() + defer r.mu.Unlock() + r.spans = append(r.spans, observedSpan{tenant, traceID, spanID, parentSpanID, service}) +} + +// assertObservedPair asserts the observer saw both the parent (source) and the +// child (target) spans, and that the child's parent links to the source span — +// i.e. enough information to form the source→target CALLS edge pre-sample. +func (r *recorder) assertObservedPair(t *testing.T, sourceService, targetService string) { + t.Helper() + r.mu.Lock() + defer r.mu.Unlock() + + bySpanID := make(map[string]observedSpan, len(r.spans)) + for _, s := range r.spans { + bySpanID[s.spanID] = s + } + + var sawSource, sawLinkedTarget bool + for _, s := range r.spans { + if s.service == sourceService && s.parentSpanID == "" { + sawSource = true + } + if s.service == targetService && s.parentSpanID != "" { + if parent, ok := bySpanID[s.parentSpanID]; ok && parent.service == sourceService { + sawLinkedTarget = true + } + } + } + if !sawSource { + t.Fatalf("observer never saw the %s source span (observed=%+v)", sourceService, r.spans) + } + if !sawLinkedTarget { + t.Fatalf("observer never saw the %s child linked to its %s parent (observed=%+v)", targetService, sourceService, r.spans) + } +} diff --git a/main.go b/main.go index d7bcbb6..95bb72e 100644 --- a/main.go +++ b/main.go @@ -571,6 +571,12 @@ func main() { graphRAG.OnSpanIngested(span) }) + // Observe cross-service call topology pre-sample so the service map keeps + // flow direction even when sampling drops the spans forming each edge. + traceServer.SetTopologyObserver(func(tenant, traceID, spanID, parentSpanID, service string) { + graphRAG.ObserveSpanTopology(tenant, traceID, spanID, parentSpanID, service) + }) + metricsServer.SetMetricCallback(func(m tsdb.RawMetric) { eventHub.BroadcastMetric(realtime.MetricEntry{ Name: m.Name, From c39630ccc8061753428720f1ca1766c143ba6718 Mon Sep 17 00:00:00 2001 From: Amit Kumar Date: Thu, 18 Jun 2026 10:04:26 +0000 Subject: [PATCH 2/2] feat(ui): sunflower service-map layout + directional edges + stronger selection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the dagre layered layout with the deterministic phyllotaxis ("sunflower") layout. At 120+ services a layered DAG collapsed sparse/ disconnected graphs into a single ~8000px-tall column (all nodes land in dagre rank 0 under rankdir:LR), which fitView clamped to minZoom — so the dense view was an unreadable vertical line of status dots. The sunflower fills a disc evenly from 7 to 120+ services (most-critical at the centre), so the map stays a navigable disc at scale. - Wire the existing, tested radialLayout (layoutRadial) into FlowMap; edges become straight chords pinned to node borders at chip zoom and centre-to- centre at dot zoom (no routing waypoints). - Direction now rides a static ARROWHEAD (markerEnd) on every edge, coloured to match its stroke — legible at any zoom and under prefers-reduced-motion. Previously the marching-dash animation was the only cue, and it vanishes when zoomed out and for reduced-motion users. - Selection + blast radius now emphasise EDGES too: the active path (1-hop of the selection, or both endpoints in the impact cone) is accent-coloured, thickened, opaque and animated; off-path edges dim to a faint hairline. Neighbour and in-cone NODES gain an accent ring (previously unstyled); selected/root rings are stronger. - Remove the now-dead dagre engine (dagreLayout.ts) and the @dagrejs/dagre dependency; the ConstellationHome lazy chunk drops ~13 KB gz (72->60), budget cap tightened 76->66. Pairs with the backend topology-observer fix (b54f04e) so cross-service edges exist to draw even at low sampling. Verified: tsc, eslint, 339 vitest tests, budgets, jscpd 0.42%, npm audit 0 — FlowMap regression suite unchanged. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01LDQVJrixs2nJoea67a8pEG --- ui/budgets.json | 4 +- ui/package-lock.json | 16 --- ui/package.json | 1 - ui/src/components/map/FlowMap.module.css | 38 ++++- ui/src/components/map/FlowMap.tsx | 171 +++++++++++------------ ui/src/lib/dagreLayout.ts | 66 --------- 6 files changed, 115 insertions(+), 181 deletions(-) delete mode 100644 ui/src/lib/dagreLayout.ts diff --git a/ui/budgets.json b/ui/budgets.json index c30fa82..b5fc43f 100644 --- a/ui/budgets.json +++ b/ui/budgets.json @@ -1,12 +1,12 @@ { - "$comment": "Gzipped size budgets enforced by scripts/check-budgets.mjs (npm run check-budgets). 'initial' = assets referenced from dist/index.html (entry script + modulepreloads + stylesheets). Fonts are raw woff2 bytes (already compressed). lazyChunkExceptions maps a chunk-name prefix (before the content hash) to a per-chunk cap. The ConstellationHome lazy chunk carries the service map: React Flow (@xyflow/react, MIT) + the dagre layout engine (@dagrejs/dagre + graphlib, MIT — crossing-minimised layout + edge routing) computed synchronously. All lazy — initial JS is unaffected (~103 KB). Measured ConstellationHome chunk ~72 KB gz; cap set to actual + headroom.", + "$comment": "Gzipped size budgets enforced by scripts/check-budgets.mjs (npm run check-budgets). 'initial' = assets referenced from dist/index.html (entry script + modulepreloads + stylesheets). Fonts are raw woff2 bytes (already compressed). lazyChunkExceptions maps a chunk-name prefix (before the content hash) to a per-chunk cap. The ConstellationHome lazy chunk carries the service map: React Flow (@xyflow/react, MIT) + our deterministic phyllotaxis 'sunflower' layout (radialLayout.ts, hand-rolled, no layout dep) computed synchronously. All lazy — initial JS is unaffected (~103 KB). Measured ConstellationHome chunk ~60 KB gz (dropped ~13 KB when @dagrejs/dagre was removed); cap set to actual + headroom.", "gzipKb": { "initialJs": 118, "initialCss": 6, "fonts": 75, "lazyChunkDefault": 10, "lazyChunkExceptions": { - "ConstellationHome": 76 + "ConstellationHome": 66 } } } diff --git a/ui/package-lock.json b/ui/package-lock.json index 30cae65..cce000a 100644 --- a/ui/package-lock.json +++ b/ui/package-lock.json @@ -8,7 +8,6 @@ "name": "otelcontext-ui", "version": "0.0.0", "dependencies": { - "@dagrejs/dagre": "^3.0.0", "@radix-ui/react-dialog": "^1.1.16", "@radix-ui/react-dropdown-menu": "^2.1.17", "@radix-ui/react-tabs": "^1.1.14", @@ -514,21 +513,6 @@ "node": ">=20.19.0" } }, - "node_modules/@dagrejs/dagre": { - "version": "3.0.0", - "resolved": "https://registry.npmjs.org/@dagrejs/dagre/-/dagre-3.0.0.tgz", - "integrity": "sha512-ZzhnTy1rfuoew9Ez3EIw4L2znPGnYYhfn8vc9c4oB8iw6QAsszbiU0vRhlxWPFnmmNSFAkrYeF1PhM5m4lAN0Q==", - "license": "MIT", - "dependencies": { - "@dagrejs/graphlib": "4.0.1" - } - }, - "node_modules/@dagrejs/graphlib": { - "version": "4.0.1", - "resolved": "https://registry.npmjs.org/@dagrejs/graphlib/-/graphlib-4.0.1.tgz", - "integrity": "sha512-IvcV6FduIIAmLwnH+yun+QtV36SC7mERqa86aClNqmMN09WhmPPYU8ckHrZBozErf+UvHPWOTJYaGYiIcs0DgA==", - "license": "MIT" - }, "node_modules/@emnapi/core": { "version": "1.10.0", "resolved": "https://registry.npmjs.org/@emnapi/core/-/core-1.10.0.tgz", diff --git a/ui/package.json b/ui/package.json index a14e38e..138e620 100644 --- a/ui/package.json +++ b/ui/package.json @@ -13,7 +13,6 @@ "check-budgets": "node scripts/check-budgets.mjs" }, "dependencies": { - "@dagrejs/dagre": "^3.0.0", "@radix-ui/react-dialog": "^1.1.16", "@radix-ui/react-dropdown-menu": "^2.1.17", "@radix-ui/react-tabs": "^1.1.14", diff --git a/ui/src/components/map/FlowMap.module.css b/ui/src/components/map/FlowMap.module.css index c72b926..29f0ff8 100644 --- a/ui/src/components/map/FlowMap.module.css +++ b/ui/src/components/map/FlowMap.module.css @@ -53,9 +53,12 @@ * reduced motion (dur 0) preserves the static accent end-state. */ .selected { border-color: var(--accent); - box-shadow: 0 0 0 1px var(--accent); - transform: scale(1.015); + box-shadow: + 0 0 0 2px var(--accent), + 0 0 0 6px var(--accent-muted); + transform: scale(1.03); transform-origin: center; + z-index: 2; } /* Field-wide focus-distortion: non-neighbour fade. */ @@ -66,13 +69,28 @@ /* Blast-radius root (?impact=): same accent emphasis as the inspected node. */ .impactRoot { border-color: var(--accent); - box-shadow: 0 0 0 1px var(--accent); + box-shadow: + 0 0 0 2px var(--accent), + 0 0 0 6px var(--accent-muted); + z-index: 2; +} + +/* 1-hop neighbour of the selected node, and intermediate blast-radius cone + * members: lift above the dimmed field with an accent outline at full opacity so + * the active neighbourhood / cone reads as a connected group, not stragglers. */ +.neighbor, +.inCone { + border-color: var(--accent); + box-shadow: 0 0 0 1px var(--accent-edge); + opacity: 1; + z-index: 1; } /* ---- level-of-detail: a status dot replaces the chip when zoomed out (dense - * 120-service view), so the 168px boxes stop occluding the routed edges. The - * dot is centred in the node box (dagre's edge waypoints still meet the box - * border) and carries the status hue. ---- */ + * 120-service view), so the 168px boxes stop occluding the edges. In dot mode + * edges connect centre-to-centre so they meet the dot rather than ending at the + * wrapper-box border; the dot is centred in the node box and carries the status + * hue. ---- */ .dotNode { display: flex; align-items: center; @@ -97,6 +115,14 @@ 0 0 0 4px var(--accent); } +/* Neighbour / in-cone dot: a thinner accent ring — present but subordinate to + * the selected/root dot's wider ring. */ +.dotNeighbor::before { + box-shadow: + 0 0 0 2px var(--bg-inset), + 0 0 0 3px var(--accent); +} + /* Connection anchor, kept INVISIBLE. React Flow needs a source/target handle * per node, but the floating edge computes the real border attachment point, so * the handle dot itself must not show — otherwise edgeless nodes sprout a stray diff --git a/ui/src/components/map/FlowMap.tsx b/ui/src/components/map/FlowMap.tsx index 3b1d0a9..5205330 100644 --- a/ui/src/components/map/FlowMap.tsx +++ b/ui/src/components/map/FlowMap.tsx @@ -14,6 +14,7 @@ import { BaseEdge, Controls, Handle, + MarkerType, MiniMap, Position, ReactFlow, @@ -32,22 +33,25 @@ import { } from '@xyflow/react' import '@xyflow/react/dist/style.css' import { NODE_H, NODE_W, compareIds, edgeWidth, neighborsOf, type GraphEdgeRef } from '@/lib/dagLayout' -import { layoutDagre } from '@/lib/dagreLayout' +import { layoutRadial } from '@/lib/radialLayout' import { formatPercent } from '@/lib/format' import { nodeStatus, statusToken } from '@/lib/triage' import type { SystemEdge, SystemNode } from '@/types/api' import styles from './FlowMap.module.css' // React Flow service map. React Flow owns pan / zoom / fit, the grid, the -// minimap and the zoom controls; we feed it a dagre layout (dagreLayout.ts — -// crossing-minimised, left→right) for node positions AND edge ROUTING -// waypoints, so links thread between ranks instead of disappearing under node -// boxes at ~120 services. Two readability levers for scale: edges are drawn -// through the dagre waypoints (routed, capped-thin), and nodes drop to status -// DOTS below a zoom threshold (level-of-detail) so the 168px chips stop -// occluding the field when zoomed out. dagre seeds the positions; nodes stay -// DRAGGABLE for manual nudges — a drag survives metric polls, and a fresh -// layout (topology change) re-seeds. +// minimap and the zoom controls; we feed it a deterministic phyllotaxis +// ("sunflower") layout (radialLayout.ts) for node positions — the most-critical +// service near the centre, the rest on a golden-angle spiral that fills the disc +// EVENLY from 7 to 120+ services. Unlike a layered DAG, a sunflower never +// collapses sparse/disconnected graphs into a single column, so the map stays a +// readable disc at scale instead of a 8000px-tall line. Two readability levers: +// edges carry a direction ARROWHEAD (markerEnd) so flow is legible statically — +// at any zoom and under reduced motion — and nodes drop to status DOTS below a +// zoom threshold (level-of-detail) so the 168px chips stop occluding the field +// when zoomed out. The layout seeds the positions; nodes stay DRAGGABLE for +// manual nudges — a drag survives metric polls, and a fresh layout (topology +// change) re-seeds. export type LayoutMode = 'radial' | 'dag' @@ -59,7 +63,7 @@ export interface FlowMapHandle { interface FlowMapProps { nodes: readonly SystemNode[] edges: readonly SystemEdge[] - /** Layout paradigm. Only the layered dagre layout is used today; kept for API stability. */ + /** Layout paradigm. The sunflower (phyllotaxis) layout is used today; kept for API stability. */ mode?: LayoutMode /** The inspected service — accent ring + 1-hop neighborhood emphasis. */ selectedId: string | null @@ -116,6 +120,7 @@ function ServiceMapNode({ data }: NodeProps) { const cls = [ styles.dotNode, emphasis === 'dim' && styles.dimmed, + (emphasis === 'neighbor' || (impactDepth !== null && impactDepth > 0)) && styles.dotNeighbor, (emphasis === 'selected' || impactDepth === 0) && styles.dotSelected, ] .filter(Boolean) @@ -129,7 +134,9 @@ function ServiceMapNode({ data }: NodeProps) { const cls = [ styles.chip, emphasis === 'selected' && styles.selected, + emphasis === 'neighbor' && styles.neighbor, emphasis === 'dim' && styles.dimmed, + impactDepth !== null && impactDepth > 0 && styles.inCone, impactDepth === 0 && styles.impactRoot, ] .filter(Boolean) @@ -147,32 +154,6 @@ function ServiceMapNode({ data }: NodeProps) { const NODE_TYPES: NodeTypes = { service: ServiceMapNode } -/** Unit vector (guards zero length). */ -function unit(dx: number, dy: number): { x: number; y: number } { - const m = Math.hypot(dx, dy) || 1 - return { x: dx / m, y: dy / m } -} - -/** SVG path through the dagre waypoints with lightly rounded corners — the - * routed "circuit" look, radius clamped so it never overshoots a short leg. */ -function routedPath(pts: { x: number; y: number }[], radius = 7): string { - if (pts.length < 2) return '' - if (pts.length === 2) return `M${pts[0].x},${pts[0].y} L${pts[1].x},${pts[1].y}` - let d = `M${pts[0].x},${pts[0].y}` - for (let i = 1; i < pts.length - 1; i++) { - const p0 = pts[i - 1] - const p1 = pts[i] - const p2 = pts[i + 1] - const r = Math.min(radius, Math.hypot(p1.x - p0.x, p1.y - p0.y) / 2, Math.hypot(p2.x - p1.x, p2.y - p1.y) / 2) - const u1 = unit(p1.x - p0.x, p1.y - p0.y) - const u2 = unit(p2.x - p1.x, p2.y - p1.y) - d += ` L${p1.x - u1.x * r},${p1.y - u1.y * r} Q${p1.x},${p1.y} ${p1.x + u2.x * r},${p1.y + u2.y * r}` - } - const last = pts.at(-1)! - d += ` L${last.x},${last.y}` - return d -} - /** Centre of an internal node in flow coords (measured dims, NODE_* fallback). */ function nodeCenter(n: InternalNode): { x: number; y: number } { return { @@ -195,31 +176,28 @@ function borderToward(n: InternalNode, toward: { x: number; y: number }): { x: n return { x: c.x + ex / scale, y: c.y + ey / scale } } -/** Edge drawn along dagre's routing waypoints (carried in data.points) so it - * threads between ranks at chip zoom. Two cases fall back to a centre-to-centre - * line instead: (1) LOD/dot zoom — chips collapse to a centred dot and the - * routed path ends ~84px away at the box border, so it would read as - * disconnected; (2) the cheap fallback layout, which has no waypoints. Either - * way the link visibly joins the two nodes. Colour/thickness ride the forwarded - * style; the flow animation is React Flow's native `animated` flag. */ -function RoutedEdge({ source, target, data, style }: EdgeProps) { +/** Straight edge between two nodes carrying a direction ARROWHEAD. The + * phyllotaxis layout has no routing waypoints, so edges are chords across the + * disc. At chip zoom each endpoint pins to the node BORDER facing the other + * node, so the arrowhead lands on the chip edge and follows a dragged node. At + * LOD/dot zoom the chips collapse to a centred dot — the 168×40 wrapper border + * would leave the line ~75px short of the dot — so we connect centre-to-centre + * instead. Colour/thickness ride the forwarded style; markerEnd is the + * forwarded arrowhead; the flow animation is React Flow's native `animated` + * flag (a secondary cue — the arrowhead is the load-bearing one). */ +function RoutedEdge({ source, target, style, markerEnd }: EdgeProps) { const compact = useStore((s) => s.transform[2] < LOD_ZOOM) const s = useInternalNode(source) const t = useInternalNode(target) if (!s || !t) return null - const pts = (data?.points as { x: number; y: number }[] | undefined) ?? [] - // Routed (chip zoom, with waypoints): keep dagre's middle routing, but pin the - // endpoints to each node's BORDER toward the adjacent waypoint — so links meet - // the chip edge (not its centre) and still follow a dragged node. - if (!compact && pts.length >= 2) { - const a = borderToward(s, pts[1] ?? nodeCenter(t)) - const b = borderToward(t, pts.at(-2) ?? nodeCenter(s)) - return + if (compact) { + const a = nodeCenter(s) + const b = nodeCenter(t) + return } - // LOD dot zoom, or the fallback layout with no waypoints: plain centre line. - const a = nodeCenter(s) - const b = nodeCenter(t) - return + const a = borderToward(s, nodeCenter(t)) + const b = borderToward(t, nodeCenter(s)) + return } const EDGE_TYPES: EdgeTypes = { routed: RoutedEdge } @@ -258,20 +236,19 @@ function FlowMapInner({ [edges], ) - // Layout: recomputed ONLY when the node/edge SET changes (deterministic - // dagre). Carries both node positions and edge routing waypoints. + // Layout: recomputed ONLY when the node/edge SET changes — the phyllotaxis + // layout is deterministic for a given node/edge set. const shapeKey = useMemo(() => { const ids = nodes.map((n) => n.id).sort(compareIds) const es = edgeRefs.map((e) => `${e.source}->${e.target}`).sort((a, b) => a.localeCompare(b)) return JSON.stringify({ ids, es }) }, [nodes, edgeRefs]) - // dagre layout, computed synchronously. It's deterministic and instant at - // normal scale; the previous Web Worker added cross-browser fragility for a - // benefit (avoiding a ~0.5–1.2s freeze only at ~120 services) that doesn't - // apply here, so it was removed for this simpler, robust path. + // Phyllotaxis ("sunflower") layout, computed synchronously — deterministic and + // instant. Edges feed criticality ranking (the most-called service lands at + // the disc centre); positions are stable for a given node/edge set. const layout = useMemo(() => { const { ids } = JSON.parse(shapeKey) as { ids: string[] } - return layoutDagre(ids, edgeRefs) + return layoutRadial(ids, edgeRefs) // eslint-disable-next-line react-hooks/exhaustive-deps }, [shapeKey]) @@ -331,32 +308,46 @@ function FlowMapInner({ return () => cancelAnimationFrame(raf) }, [layout, fitView]) - // Routed animated edges. Waypoints come from the dagre layout; token colour - // rides the inline style (CSS custom props resolve in the style property, - // unlike SVG presentation attributes); width is capped so high call-counts - // stay a hairline. Failing edges carry --crit + an `edgeFailing` class hook. - const rfEdges: Edge[] = useMemo( - () => - edges.map((e) => { - const failing = isEdgeFailing(e.status) - const key = `${e.source}->${e.target}` - return { - id: key, - source: e.source, - target: e.target, - type: 'routed', - animated: true, - className: failing ? 'edgeFailing' : undefined, - data: { points: layout.edges.get(key) ?? [] }, - style: { - stroke: failing ? 'var(--crit)' : 'var(--text-3)', - strokeWidth: Math.min(2.25, edgeWidth(e.call_count)), - opacity: failing ? 1 : 0.7, - } as CSSProperties, - } - }), - [edges, layout], - ) + // Selection-aware animated edges. With a node selected (or a blast-radius + // cone) the ACTIVE path — 1-hop edges of the selection, or cone edges whose + // BOTH endpoints are in the cone — is drawn in the accent hue, thicker, fully + // opaque and animated; every off-path edge recedes to a dim hairline and stops + // animating, so the path reads as a clear through-line. A failing edge keeps + // its --crit hue. With nothing selected it's the neutral resting style. Token + // colour rides the inline style (CSS custom props resolve there, unlike SVG + // presentation attributes); width is capped so high call-counts stay a + // hairline. Every edge carries a direction arrowhead (markerEnd) coloured to + // match its stroke, so flow direction is legible statically — at any zoom and + // under reduced motion. Failing edges keep the `edgeFailing` class hook. + const rfEdges: Edge[] = useMemo(() => { + const focusing = selectedId !== null || impact != null + return edges.map((e) => { + const failing = isEdgeFailing(e.status) + const key = `${e.source}->${e.target}` + const active = impact + ? impact.has(e.source) && impact.has(e.target) + : selectedId !== null && (e.source === selectedId || e.target === selectedId) + const dimmed = focusing && !active + const stroke = failing ? 'var(--crit)' : active ? 'var(--accent)' : 'var(--text-3)' + const width = active + ? Math.min(3.5, edgeWidth(e.call_count) * 1.6) + : Math.min(2.25, edgeWidth(e.call_count)) + return { + id: key, + source: e.source, + target: e.target, + type: 'routed', + animated: !dimmed, + className: failing ? 'edgeFailing' : undefined, + markerEnd: { type: MarkerType.ArrowClosed, width: 16, height: 16, color: stroke }, + style: { + stroke, + strokeWidth: dimmed ? 1 : width, + opacity: dimmed ? 0.14 : failing || active ? 1 : 0.7, + } as CSSProperties, + } + }) + }, [edges, selectedId, impact]) const onNodeClick = useCallback((_: unknown, n: Node) => onSelect(n.id), [onSelect]) diff --git a/ui/src/lib/dagreLayout.ts b/ui/src/lib/dagreLayout.ts deleted file mode 100644 index 6941cfa..0000000 --- a/ui/src/lib/dagreLayout.ts +++ /dev/null @@ -1,66 +0,0 @@ -// Layered service-map layout via dagre (@dagrejs/dagre, MIT). Unlike our -// hand-rolled dagLayout, dagre does crossing-minimisation AND emits edge -// ROUTING waypoints (g.edge().points) — polylines that thread between ranks -// instead of cutting straight through node boxes. That routing is what keeps -// edges visible once the graph grows to ~120 services. Left→right, deterministic -// (ids + edges sorted before insertion, dagre is stable for fixed input order). - -import dagre from '@dagrejs/dagre' -import { NODE_H, NODE_W, compareIds, type GraphEdgeRef } from './dagLayout' - -export interface DagreLayout { - /** id → node top-left position in React Flow flow coords. */ - nodes: Map - /** "source->target" → routed polyline waypoints (same flow coords). */ - edges: Map - width: number - height: number -} - -const RANKSEP = 110 // gap between dependency ranks (the long axis, LR) -const NODESEP = 26 // gap between siblings within a rank -const EDGESEP = 14 // gap between parallel edges in a routing channel - -const EMPTY: DagreLayout = { nodes: new Map(), edges: new Map(), width: 0, height: 0 } - -/** Deterministic left→right layered layout with edge routing. */ -export function layoutDagre(nodeIds: readonly string[], edges: readonly GraphEdgeRef[]): DagreLayout { - const ids = [...new Set(nodeIds)].sort(compareIds) - if (ids.length === 0) return EMPTY - - const g = new dagre.graphlib.Graph() - g.setGraph({ rankdir: 'LR', nodesep: NODESEP, ranksep: RANKSEP, edgesep: EDGESEP, marginx: NODE_W / 2, marginy: NODE_H }) - g.setDefaultEdgeLabel(() => ({})) - for (const id of ids) g.setNode(id, { width: NODE_W, height: NODE_H }) - - // Sanitize → dedupe → sort for a deterministic build. - const idSet = new Set(ids) - const seen = new Set() - const clean: GraphEdgeRef[] = [] - for (const e of edges) { - if (e.source === e.target || !idSet.has(e.source) || !idSet.has(e.target)) continue - const k = `${e.source}->${e.target}` - if (seen.has(k)) continue - seen.add(k) - clean.push(e) - } - clean.sort((a, b) => compareIds(`${a.source}->${a.target}`, `${b.source}->${b.target}`)) - for (const e of clean) g.setEdge(e.source, e.target) - - dagre.layout(g) - - const nodes = new Map() - for (const id of ids) { - const n = g.node(id) - if (n) nodes.set(id, { x: n.x - NODE_W / 2, y: n.y - NODE_H / 2 }) - } - const edgeMap = new Map() - for (const e of clean) { - const ed = g.edge(e.source, e.target) - if (ed && Array.isArray(ed.points)) { - edgeMap.set(`${e.source}->${e.target}`, ed.points.map((p: { x: number; y: number }) => ({ x: p.x, y: p.y }))) - } - } - const gr = g.graph() - return { nodes, edges: edgeMap, width: gr.width ?? 0, height: gr.height ?? 0 } -}