package indexer import ( "context" "sort" "strings" "github.com/zzet/gortex/internal/clones" "github.com/zzet/gortex/internal/graph" "github.com/zzet/gortex/internal/parser" "github.com/zzet/gortex/internal/progress" ) // cloneSigMetaKey is the Node.Meta key under which a function/method's // base64-encoded MinHash signature is stored. The graph-wide LSH pass // reads it back out — keeping the signature on the node makes the pass // a pure graph walk (no file IO), correct under incremental reindex, // and safe across multi-repo graphs. const cloneSigMetaKey = "clone_sig" // cloneTokensMetaKey is the Node.Meta key under which the normalised- // token count of a function/method body is stored alongside the clone // signature. Used by the length-stratified LSH pass to bucket items // into overlapping size classes so a pair with size ratio > ~1.6 // (Jaccard ≤ 0.625, well below the 0.82 clone threshold) is never // considered as a candidate. const cloneTokensMetaKey = "clone_tokens" // cloneShinglesMetaKey is the Node.Meta key under which a function / // method's raw shingle hash set is stashed during the per-file parse, // so the global CMS-filter pass (finaliseCloneSignatures) can decide // which shingles to exclude before computing the final MinHash // signature. The entry is deleted from Meta as soon as the signature // lands — it is intentionally short-lived because the shingle set is // large (≈ tokens − 2 entries per body) and persisting it across the // clone-detection pass would waste tens of MB on a monorepo. const cloneShinglesMetaKey = "clone_shingles" // CMS-filter tuning. // // cmsBoilerplateRatio: a shingle appearing in more than this fraction // of bodies is treated as boilerplate and excluded from signature // computation. 1% is the textbook value used by near-duplicate web // indexing systems and balances precision (false-clone suppression) // against recall (genuine clones whose shared content happens to use // a moderately common idiom). // // cmsMinCorpus: below this many bodies the global frequency // distribution is too thin for the threshold to be meaningful — a // 200-body repo has no shingle that legitimately appears in 2 bodies // without already being noise — so we fall back to unfiltered MinHash. // Around this size the LSH pass is also fast enough that filtering // gains nothing. // // minSurvivingShingles: after filtering, a body with fewer // discriminative shingles than this is dropped from clone detection // entirely. MinHash over a handful of shingles produces random slot // values that collide unpredictably in LSH bands; the body is then a // false-clone factory, not a real clone source. Boilerplate-dominated // bodies (e.g. trivial controller / DTO wrappers) land here. const ( cmsBoilerplateRatio = 0.01 minSurvivingShingles = 8 ) // cmsMinCorpus is the body-count floor below which the CMS boilerplate // filter is disabled (useFilter=false) and the pass falls back to // unfiltered MinHash — see the doc comment above for the rationale and // default. It is a package-level var (not a const) purely so the clone // equivalence tests can temporarily lower it to force useFilter=true on a // small fixture and exercise the filtered batch/incremental paths; restore // it via t.Cleanup. Production never mutates it — the default semantics are // unchanged. var cmsMinCorpus = 2000 // applyCloneSignatures is the per-file half of clone detection. It runs // inside applyCoverageDomains (gated on the "clones" coverage domain), // slices each function/method body out of the file source, computes a // MinHash signature, and stamps it on the node's Meta. Bodies below // clones.MinTokens normalised tokens produce no signature and are // silently skipped — they are dominated by boilerplate and would only // add noise to the LSH buckets. // // Allocation note: the body slicing path computes one []int of line // offsets per file and one string per emitted body. The previous // implementation went through splitLines (which materialises the // whole source as N per-line Go strings) and a quadratic concat in // bodyText (each iteration grew the output via "out += ..."). Profile // showed bodyText + splitLinesUpTo at 3+ GiB per 30 s window — both // are now O(file_bytes) one-shot allocations. func applyCloneSignatures(src []byte, result *parser.ExtractionResult) { if result == nil || len(result.Nodes) == 0 { return } // Compute newline offsets once per file rather than splitting the // source into N Go strings. offsets[i] is the byte index where // line i+1 (1-indexed) starts; the sentinel offsets[len(offsets)-1] // is len(src) so the slice math doesn't need a special case for // the last line. offsets := lineOffsets(src) for _, n := range result.Nodes { if n == nil { continue } if n.Kind != graph.KindFunction && n.Kind != graph.KindMethod { continue } body := bodyTextFromOffsets(src, offsets, n.StartLine, n.EndLine) if body == "" { continue } // Stash the deduplicated shingle set rather than the final // MinHash signature: signature computation is deferred to the // global CMS-filter pass (finaliseCloneSignatures), which // derives a per-corpus boilerplate-shingle set and excludes it // from each body's signature. The shingle slice is short-lived // on Meta — finaliseCloneSignatures clears it after stamping // the real signature. shingles, tokens, ok := clones.Shingles(body) if !ok { continue } if n.Meta == nil { n.Meta = map[string]any{} } n.Meta[cloneShinglesMetaKey] = shingles n.Meta[cloneTokensMetaKey] = tokens } } // lineOffsets returns the byte offsets of each line in src. For a file // with N lines the result has length N+1: the first entry is 0, each // subsequent entry is the byte index immediately after a '\n', and the // final sentinel is len(src) so callers can slice the last line as // src[offsets[N-1]:offsets[N]] without special-casing EOF. // // One allocation (the []int) instead of N (one string per line via // strings.Split). Lifetime is per-file: the caller drops the slice // when the file's worker batch finishes. func lineOffsets(src []byte) []int { // Reserve a generous initial capacity to avoid repeated slice // growth on typical source files (~ 200 lines). The slice grows // from here for larger files; small files waste a bit of headroom // that goes back to the GC immediately. offsets := make([]int, 1, 256) for i := 0; i < len(src); i++ { if src[i] == '\n' { offsets = append(offsets, i+1) } } offsets = append(offsets, len(src)) return offsets } // bodyTextFromOffsets returns src[startLine..endLine] (both 1-indexed, // inclusive) as one Go string. The trailing newline of the last // included line is stripped so output matches the old line-join // semantics ("a\nb" not "a\nb\n"). Returns "" for degenerate or // out-of-bounds ranges, matching bodyText. func bodyTextFromOffsets(src []byte, offsets []int, startLine, endLine int) string { if startLine <= 0 || endLine < startLine { return "" } lo := startLine - 1 hi := endLine // len(offsets) = lineCount + 1 (sentinel). lineCount = len(offsets) - 1. lineCount := len(offsets) - 1 if lo >= lineCount { return "" } if hi > lineCount { hi = lineCount } startOff := offsets[lo] endOff := offsets[hi] // Strip the trailing '\n' that bounds the last included line so the // output matches the line-join semantics callers and tests expect. if endOff > startOff && endOff <= len(src) && endOff-1 >= 0 && src[endOff-1] == '\n' { endOff-- } return string(src[startOff:endOff]) } // bodyText returns the source spanning [startLine, endLine] (both // 1-indexed, inclusive) joined by newlines. Kept as a legacy helper // for the unit-test surface; production callers go through // applyCloneSignatures → bodyTextFromOffsets, which avoids both the // whole-source string copy in splitLines and the O(N²) concat below. func bodyText(lines []string, startLine, endLine int) string { if startLine <= 0 || endLine < startLine { return "" } lo := startLine - 1 hi := endLine if lo >= len(lines) { return "" } if hi > len(lines) { hi = len(lines) } // Precompute the joined size so the strings.Builder grows once, // turning the previous O(N²) "out += ..." into O(total_bytes). total := 0 for i := lo; i < hi; i++ { total += len(lines[i]) if i > lo { total++ // separating '\n' } } var b strings.Builder b.Grow(total) for i := lo; i < hi; i++ { if i > lo { b.WriteByte('\n') } b.WriteString(lines[i]) } return b.String() } // computeCloneSigFromShingles is the per-body signature kernel shared by // the whole-graph finalise pass (finaliseCloneSignatures) and the // incremental maintainer (incrementalCloneIndex.UpdateFuncs). Both paths // MUST route through this function so a body's signature is byte-identical // regardless of which path stamped it — that is what lets the equivalence // test assert exact set equality between the batch and incremental clone // edges. // // cms is the corpus Count-Min Sketch; threshold is the boilerplate cutoff // (a shingle whose CMS count exceeds it is dropped). useFilter selects the // branch: // // - useFilter true: exclude high-frequency shingles, then require the // surviving set to clear minSurvivingShingles before computing MinHash. // - useFilter false: keep every shingle and apply no floor (legacy // small-corpus behaviour) — cms may be nil in this branch. // // Returns the signature and ok=false when the body is dropped from clone // detection (empty / below the surviving floor) — the caller then leaves // the node without a clone_sig, exactly as the batch pass does. func computeCloneSigFromShingles(cms *clones.CMS, threshold uint32, useFilter bool, shingles []uint64) (clones.Signature, bool) { var filtered []uint64 if useFilter { filtered = make([]uint64, 0, len(shingles)) for _, sh := range shingles { if cms.Count(sh) > threshold { continue } filtered = append(filtered, sh) } } else { filtered = shingles } floor := minSurvivingShingles if !useFilter { // Without filtering, every shingle survives — fall back to the // legacy gate so we don't silently drop bodies the old code // would have kept. floor = 0 } return clones.SignatureFromShingles(filtered, floor) } // finaliseCloneSignatures runs after every file's shingles have been // stamped on its function / method nodes (by applyCloneSignatures // during the per-file parse). It builds a Count-Min Sketch of shingle // frequencies across every body in the graph, then walks the bodies // again and computes a MinHash signature excluding shingles that // exceed the boilerplate threshold (present in > cmsBoilerplateRatio // of bodies). The stashed shingle set is cleared from Meta as soon as // the signature lands so the LSH pass downstream sees the same // node-shape the legacy path produced — just with cleaner signatures. // // Bodies whose surviving shingle count falls below minSurvivingShingles // are dropped from clone detection entirely (no clone_sig stamp): a // body whose token stream is dominated by boilerplate is, by // definition, a controller / DTO / dispatch shape rather than // distinguishable code, and including it in MinHash would just produce // random LSH collisions. // // Below cmsMinCorpus bodies the corpus is too small for the // frequency distribution to be meaningful; the pass falls back to // unfiltered MinHash so small repos preserve the legacy behaviour. // // Caller must hold g.ResolveMutex() — the function mutates Node.Meta // (deletes clone_shingles, sets clone_sig) across nodes that other // graph-wide passes (markTestSymbolsAndEmitEdges, ResolveTemporalCalls, // reach.BuildIndex) also touch under the same mutex. // // Repo-scoped: only bodies whose n.RepoPrefix == repoPrefix enter the // CMS / signature passes, so a multi-repo graph computes each repo's // boilerplate sketch and per-body signatures from that repo's bodies // alone — clone detection is per-repository. A standalone single-repo // Indexer uses repoPrefix == "" and its nodes carry RepoPrefix == "", // so the equality matches every node and behaviour is unchanged. // (GetRepoNodes can't be used here: GetRepoNodes("") is empty for the // in-memory / single-repo store — see incrementalCloneIndex.Rebuild — // so the AllNodes + equality filter is the form that works for both // regimes, since "" == "" matches every node.) // cloneRepoNodes returns the nodes the per-repo clone passes must walk. In // daemon multi-repo mode repoPrefix is non-empty, so GetRepoNodes selects just // that repo's nodes (one backend query, and one meta decode per repo node) // instead of decoding every node in a many-repo graph only to discard the other // repos' — the whole-graph AllNodes scan these passes used to run per repo. // In single-repo / in-memory mode repoPrefix is "" and those nodes are not // tracked in the per-repo buckets GetRepoNodes reads, so the AllNodes fallback // (whose "" == n.RepoPrefix filter matches every node) is the only form that // works there. Callers keep their n.RepoPrefix == repoPrefix guard: a no-op on // the GetRepoNodes path, load-bearing on the AllNodes fallback. The clone passes // read blob-only Meta (clone_sig / clone_tokens / clone_shingles), so the full // GetRepoNodes — not the meta-less light reader — is required here. func cloneRepoNodes(g graph.Store, repoPrefix string) []*graph.Node { if repoPrefix != "" { return g.GetRepoNodes(repoPrefix) } return g.AllNodes() } func finaliseCloneSignatures(g graph.Store, repoPrefix string) { // First pass: collect every body that has stashed shingles. We // capture the *graph.Node pointers up front so the CMS-build pass // and the signature-compute pass don't both re-walk g.AllNodes(). bodies := make([]*graph.Node, 0, 8192) for _, n := range cloneRepoNodes(g, repoPrefix) { if n == nil || n.Meta == nil { continue } if n.RepoPrefix != repoPrefix { continue } if n.Kind != graph.KindFunction && n.Kind != graph.KindMethod { continue } if _, ok := n.Meta[cloneShinglesMetaKey].([]uint64); !ok { continue } bodies = append(bodies, n) } if len(bodies) == 0 { return } useFilter := len(bodies) >= cmsMinCorpus var cms *clones.CMS var threshold uint32 if useFilter { // Default sketch sizing — see the CMS doc comment for the // width/depth → ε/δ derivation. 1 MB peak for a transient, // per-build pass is comfortably below any constraint. cms = clones.NewCMS(65536, 4) for _, n := range bodies { shingles, _ := n.Meta[cloneShinglesMetaKey].([]uint64) for _, sh := range shingles { cms.Add(sh) } } threshold = uint32(float64(len(bodies)) * cmsBoilerplateRatio) if threshold < 1 { threshold = 1 } } // Persist each body's raw shingle set to the clone_shingles sidecar // BEFORE deleting it from Meta. This loop walks EVERY body in the // corpus — both the survivors (which get a clone_sig below) and the // boilerplate-dropped bodies (which do not) — persisting any with a // non-empty shingle set. That is deliberate: incrementalCloneIndex. // Rebuild reseeds its CMS + corpus from these rows and must mirror // the bodies set this pass used to build its own CMS / threshold, // which is ALL eligible bodies, not just survivors. Persisting only // survivors here would under-seed Rebuild's sketch and skew the // incremental threshold away from the batch one. Meta stays lean // (the shingle set is large and only the CMS pass needs it), but the // durable sidecar copy lets a warm restart rebuild the incremental // CMS without re-parsing every body. Accumulate per node.RepoPrefix // so a multi-repo graph reseeds each repo's CMS in isolation. // Backends that don't implement CloneShingleWriter (no on-disk store) // simply skip this — the in-session incremental index caches shingles // in memory regardless. if w, ok := g.(graph.CloneShingleWriter); ok { byPrefix := make(map[string]map[string][]uint64) for _, n := range bodies { shingles, _ := n.Meta[cloneShinglesMetaKey].([]uint64) if len(shingles) == 0 { continue } rows := byPrefix[n.RepoPrefix] if rows == nil { rows = make(map[string][]uint64) byPrefix[n.RepoPrefix] = rows } rows[n.ID] = shingles } for prefix, rows := range byPrefix { _ = w.BulkSetCloneShingles(prefix, rows) } } // Second pass: signature computation. Each body either lands a // fresh clone_sig (signature over surviving shingles) or is // dropped entirely (no clone_sig, never enters detection items // list). In both cases clone_shingles is removed from Meta. The // per-body kernel is computeCloneSigFromShingles — the incremental // maintainer calls the same kernel so signatures match exactly. for _, n := range bodies { shingles, _ := n.Meta[cloneShinglesMetaKey].([]uint64) sig, ok := computeCloneSigFromShingles(cms, threshold, useFilter, shingles) delete(n.Meta, cloneShinglesMetaKey) if !ok { // Boilerplate-dominated or empty after filter — drop // from clone detection. detectClonesAndEmitEdges skips // nodes without a clone_sig. continue } n.Meta[cloneSigMetaKey] = clones.EncodeSignature(sig) } } // CloneDetectionStats summarises one detectClonesAndEmitEdges run for // the caller's logger. Exposed so the orchestrator can surface what the // per-bucket cap dropped — a high skippedBucketItems means the // workspace has a lot of templated boilerplate that LSH would have // over-fanned-out on. type CloneDetectionStats struct { Items int // function/method nodes with a signature Pairs int // detected clone pairs (after Jaccard filter) Edges int // EdgeSimilarTo emitted (≈ 2·Pairs, modulo dedup) SkippedBuckets int // LSH buckets dropped for exceeding maxBucketSize SkippedBucketItems int // total items inside the dropped buckets DiffusedPairs int // semantically-related pairs surviving threshold+cap DiffusedEdges int // EdgeSemanticallyRelated emitted (= 2·DiffusedPairs) } // detectClonesAndEmitEdges is the graph-wide half of clone detection. // It collects every function/method node carrying a clone_sig, runs // the MinHash + LSH pass over their signatures, and materialises a // symmetric pair of EdgeSimilarTo edges for each detected clone pair. // // threshold is the Jaccard similarity cutoff; pass 0 to use the // clones package default. Returns clone stats including the per-bucket // cap telemetry — the orchestrator logs that so a high skip count is // visible during warmup. // // The pass is a full recompute and is idempotent: graph.AddEdge dedupes // by edgeKey so re-emitting an unchanged pair is a no-op, and stale // edges cannot survive — when either endpoint's file is reindexed, // EvictFile removes that node's edges in both directions before this // pass re-runs. // // repoPrefix scopes the pass to one repository's nodes: every whole-graph // walk it drives (finalise, item gather, diffusion) is filtered to // n.RepoPrefix == repoPrefix so no cross-repo candidate pair is ever // formed. A standalone single-repo Indexer passes "" and its nodes carry // RepoPrefix == "", so the equality matches all nodes and the single-repo // result is unchanged. func detectClonesAndEmitEdges(g graph.Store, repoPrefix string, threshold float64) CloneDetectionStats { return detectClonesAndEmitEdgesCtx(context.Background(), g, repoPrefix, threshold) } // detectClonesAndEmitEdgesCtx is the context-aware sibling of // detectClonesAndEmitEdges. It emits sub-stage progress markers via // the reporter attached to ctx (see progress.WithReporter): clone // detection is the longest single stage on monorepo-scale graphs and // without intra-stage reporters an operator sees just one // "clone detection pass" marker followed by minutes of silence — no // way to tell finalise-signatures from LSH from edge-emission. func detectClonesAndEmitEdgesCtx(ctx context.Context, g graph.Store, repoPrefix string, threshold float64) CloneDetectionStats { var stats CloneDetectionStats if g == nil { return stats } reporter := progress.FromContext(ctx) // Serialise against other graph-wide passes that mutate Node.Meta // (markTestSymbolsAndEmitEdges, ResolveTemporalCalls, reach.BuildIndex, // releases enrichment). Without this lock, the AllNodes walk below // reads n.Meta while one of those writers mutates the same map and // the runtime aborts with "concurrent map read and map write" — the // observed daemon crash. Shares g.ResolveMutex() so all such passes // rendezvous on the same lock the resolver already uses. g.ResolveMutex().Lock() defer g.ResolveMutex().Unlock() // Finalise pending signatures: applyCloneSignatures stamped the // raw shingle set on each function/method node during the per-file // parse. This pass builds a Count-Min Sketch of corpus-wide shingle // frequencies, then computes the MinHash signature for each body // after excluding shingles whose frequency exceeds the boilerplate // threshold. The expensive LSH candidate enumeration that comes // next then runs over signatures that reflect discriminative // content only — k8s-style controller-pattern bodies stop colliding // on shared "if v err return v" / "( v . v )" shingles, which is // what drives the LSH bucket explosion at monorepo scale. // // Runs under the existing g.ResolveMutex() so the Meta mutations // (delete clone_shingles, set clone_sig) don't race the AllNodes // walk below. reporter.Report("clones: CMS-finalise signatures", 0, 0) finaliseCloneSignatures(g, repoPrefix) reporter.Report("clones: gather items", 0, 0) var items []clones.Item for _, n := range cloneRepoNodes(g, repoPrefix) { if n == nil || n.Meta == nil { continue } // Scope to this repo's nodes so no cross-repo candidate pair is // ever formed. "" matches every node (single-repo / in-memory). if n.RepoPrefix != repoPrefix { continue } if n.Kind != graph.KindFunction && n.Kind != graph.KindMethod { continue } enc, ok := n.Meta[cloneSigMetaKey].(string) if !ok || enc == "" { continue } sig, ok := clones.DecodeSignature(enc) if !ok { continue } // Read the stamped token count when present. Legacy nodes // indexed before the stamp was added simply get TokenCount=0, // which lengthClassesOf treats as "unknown" → all classes, // preserving the unstratified behaviour for them. tokens := 0 switch v := n.Meta[cloneTokensMetaKey].(type) { case int: tokens = v case int64: tokens = int(v) case float64: tokens = int(v) } items = append(items, clones.Item{ID: n.ID, Sig: sig, TokenCount: tokens}) } stats.Items = len(items) if len(items) < 2 { return stats } reporter.Report("clones: LSH + Jaccard filter", len(items), 0) detected, sb, sbi := clones.DetectPairsStratifiedWithStats(items, threshold) stats.SkippedBuckets = sb stats.SkippedBucketItems = sbi stats.Pairs = len(detected) reporter.Report("clones: emit similarity edges", len(detected), 0) directPairs := make(map[[2]string]struct{}, len(detected)) for _, p := range detected { from := g.GetNode(p.A) to := g.GetNode(p.B) if from == nil || to == nil { continue } emitSimilarEdge(g, from, to, p.Similarity) emitSimilarEdge(g, to, from, p.Similarity) stats.Edges += 2 // Record the canonicalised (A nbrs[j].score } return nbrs[i].id < nbrs[j].id }) if len(nbrs) > diffusionMaxNeighbors { adj[id] = nbrs[:diffusionMaxNeighbors] } } // For each bridge node B, every unordered pair of its neighbours // (A,C) is a candidate two-hop relation. The diffused score is the // damped product of the two clone links; when multiple bridges // connect the same (A,C) the strongest (max) bridge wins. best := make(map[[2]string]float64) bridges := make([]string, 0, len(adj)) for id := range adj { bridges = append(bridges, id) } sort.Strings(bridges) for _, b := range bridges { nbrs := adj[b] for i := range nbrs { for j := i + 1; j < len(nbrs); j++ { a, c := nbrs[i].id, nbrs[j].id if a == c { continue } key := canonicalPair(a, c) if _, isClone := directPairs[key]; isClone { continue // a direct clone — stays similar_to only } score := diffusionDamping * nbrs[i].score * nbrs[j].score if score < diffusionThreshold { continue } if score > best[key] { best[key] = score } } } } if len(best) == 0 { return 0, 0 } // Rank surviving pairs by diffused score so the global cap keeps // the strongest relations; ID tie-breaks keep the cut deterministic. type diffusedPair struct { a, c string score float64 } ranked := make([]diffusedPair, 0, len(best)) for key, score := range best { ranked = append(ranked, diffusedPair{a: key[0], c: key[1], score: score}) } sort.Slice(ranked, func(i, j int) bool { if ranked[i].score != ranked[j].score { return ranked[i].score > ranked[j].score } if ranked[i].a != ranked[j].a { return ranked[i].a < ranked[j].a } return ranked[i].c < ranked[j].c }) if len(ranked) > diffusionMaxPairs { ranked = ranked[:diffusionMaxPairs] } for _, rp := range ranked { from := g.GetNode(rp.a) to := g.GetNode(rp.c) if from == nil || to == nil { continue } emitSemanticallyRelatedEdge(g, from, to, rp.score) emitSemanticallyRelatedEdge(g, to, from, rp.score) diffusedPairs++ diffusedEdges += 2 } return diffusedPairs, diffusedEdges } // emitSimilarEdge adds one directed EdgeSimilarTo edge carrying the // estimated Jaccard similarity. The edge is anchored at the source // node's file/line for locality. Origin is ast_inferred — the // relationship is a statistical estimate over normalised tokens, not a // structural fact. func emitSimilarEdge(g graph.Store, from, to *graph.Node, similarity float64) { g.AddEdge(&graph.Edge{ From: from.ID, To: to.ID, Kind: graph.EdgeSimilarTo, FilePath: from.FilePath, Line: from.StartLine, Confidence: similarity, Origin: graph.OriginASTInferred, Meta: map[string]any{"similarity": similarity}, }) } // emitSemanticallyRelatedEdge adds one directed EdgeSemanticallyRelated // edge carrying the diffused similarity score. Like emitSimilarEdge the // edge is anchored at the source node's file/line and origin is // ast_inferred — the score is a statistical estimate over normalised // tokens, here additionally smoothed across the similarity graph. func emitSemanticallyRelatedEdge(g graph.Store, from, to *graph.Node, similarity float64) { g.AddEdge(&graph.Edge{ From: from.ID, To: to.ID, Kind: graph.EdgeSemanticallyRelated, FilePath: from.FilePath, Line: from.StartLine, Confidence: similarity, Origin: graph.OriginASTInferred, Meta: map[string]any{"similarity": similarity}, }) }