From cf651e89622743cdcd1c45f63d3135c73c09c909 Mon Sep 17 00:00:00 2001 From: tae2089 Date: Thu, 16 Jul 2026 09:33:08 +0900 Subject: [PATCH 1/2] perf: reconcile bulk updates before resolving edges --- .../ingest/incremental/deferred_edge_spool.go | 100 +++++++++ .../app/ingest/incremental/import_lookup.go | 83 ++++++++ .../app/ingest/incremental/incremental.go | 169 ++++++++++++++- internal/app/ingest/ports.go | 21 ++ .../app/ingest/resolve/import_file_index.go | 63 ++++++ internal/app/ingest/workflow/build.go | 73 +------ internal/app/ingest/workflow/indexer.go | 6 + internal/app/ingest/workflow/indexer_test.go | 198 ++++++++++++++++++ internal/app/ingest/workflow/packages.go | 41 ++++ internal/app/ingest/workflow/update.go | 184 +++++++++------- 10 files changed, 799 insertions(+), 139 deletions(-) create mode 100644 internal/app/ingest/incremental/deferred_edge_spool.go create mode 100644 internal/app/ingest/incremental/import_lookup.go create mode 100644 internal/app/ingest/resolve/import_file_index.go diff --git a/internal/app/ingest/incremental/deferred_edge_spool.go b/internal/app/ingest/incremental/deferred_edge_spool.go new file mode 100644 index 0000000..3e30336 --- /dev/null +++ b/internal/app/ingest/incremental/deferred_edge_spool.go @@ -0,0 +1,100 @@ +// @index Session-local parsed-edge spool for staged incremental reconciliation. +package incremental + +import ( + "encoding/gob" + "fmt" + "log/slog" + "os" + "path/filepath" + + "github.com/tae2089/trace" + + "github.com/tae2089/code-context-graph/internal/domain/graph" +) + +// deferredEdgeFile holds one file's parsed edges after its nodes are persisted. +// @intent retain only the edge-resolution input needed after source bytes are released. +type deferredEdgeFile struct { + FilePath string + Edges []graph.Edge +} + +// deferredEdgeRecord is one bounded disk record used between node and edge phases. +// @intent keep large staged updates independent of the total parsed edge count in memory. +type deferredEdgeRecord struct { + Files []deferredEdgeFile +} + +// deferredEdgeSpool owns temporary edge records for one staged reconciliation invocation. +// @intent preserve parsed cross-batch edges until every changed node has been applied. +type deferredEdgeSpool struct { + dir string + records []string +} + +// newDeferredEdgeSpool creates an invocation-local directory for parsed edge records. +// @intent isolate temporary staged-update data so cleanup cannot affect persistent graph state. +// @sideEffect creates a temporary directory under the operating system temp root. +func newDeferredEdgeSpool() (*deferredEdgeSpool, error) { + dir, err := os.MkdirTemp("", "ccg-deferred-edge-spool-*") + if err != nil { + return nil, trace.Wrap(err, "create deferred edge spool") + } + return &deferredEdgeSpool{dir: dir}, nil +} + +// writeRecord persists a bounded parsed-edge batch for later resolution. +// @intent defer cross-file edge resolution until all batch-local node replacements are complete. +// @sideEffect creates a gob record inside the invocation-local spool directory. +func (s *deferredEdgeSpool) writeRecord(record deferredEdgeRecord) error { + if len(record.Files) == 0 { + return nil + } + path := filepath.Join(s.dir, fmt.Sprintf("%06d.gob", len(s.records))) + file, err := os.Create(path) + if err != nil { + return trace.Wrap(err, "create deferred edge spool record") + } + encodeErr := gob.NewEncoder(file).Encode(record) + closeErr := file.Close() + if encodeErr != nil { + return trace.Wrap(encodeErr, "encode deferred edge spool record") + } + if closeErr != nil { + return trace.Wrap(closeErr, "close deferred edge spool record") + } + s.records = append(s.records, path) + return nil +} + +// readRecord streams one parsed-edge batch back into the resolution phase. +// @intent let edge resolution remain bounded by the original source batch size. +func (s *deferredEdgeSpool) readRecord(path string) (deferredEdgeRecord, error) { + var record deferredEdgeRecord + file, err := os.Open(path) + if err != nil { + return record, trace.Wrap(err, "open deferred edge spool record") + } + decodeErr := gob.NewDecoder(file).Decode(&record) + closeErr := file.Close() + if decodeErr != nil { + return record, trace.Wrap(decodeErr, "decode deferred edge spool record") + } + if closeErr != nil { + return record, trace.Wrap(closeErr, "close deferred edge spool record") + } + return record, nil +} + +// cleanup removes all temporary parsed-edge records for this invocation. +// @intent ensure successful and failed staged updates do not retain temporary source-derived data. +// @sideEffect deletes the invocation-local spool directory. +func (s *deferredEdgeSpool) cleanup(logger *slog.Logger) { + if s == nil || s.dir == "" { + return + } + if err := os.RemoveAll(s.dir); err != nil && logger != nil { + logger.Warn("cleanup deferred edge spool failed", "dir", s.dir, "error", err) + } +} diff --git a/internal/app/ingest/incremental/import_lookup.go b/internal/app/ingest/incremental/import_lookup.go new file mode 100644 index 0000000..7f102a2 --- /dev/null +++ b/internal/app/ingest/incremental/import_lookup.go @@ -0,0 +1,83 @@ +// @index Transaction-local import lookup decorator for staged incremental edge replay. +package incremental + +import ( + "context" + + "github.com/tae2089/code-context-graph/internal/app/ingest/resolve" + "github.com/tae2089/code-context-graph/internal/domain/graph" +) + +// importFileNodeLister is an optional store capability for one-shot import lookup indexing. +// @intent avoid expanding the legacy incremental Store contract for lightweight test doubles. +type importFileNodeLister interface { + ListImportFileNodes(ctx context.Context) ([]graph.Node, error) +} + +// fileSuffixLookup is the legacy import lookup capability implemented by production graph stores. +// @intent keep staged reconciliation compatible with custom stores that do not expose the bulk file-node query. +type fileSuffixLookup interface { + GetFileNodesByPathSuffix(ctx context.Context, suffix string) ([]graph.Node, error) +} + +// edgeReader preserves existing implements lookup when the decorated store supports it. +// @intent keep staged resolution behavior aligned with the underlying graph store capabilities. +type edgeReader interface { + GetEdgesToNodes(ctx context.Context, nodeIDs []uint) ([]graph.Edge, error) +} + +// importIndexedLookup shares one immutable import-file index across staged edge replay. +// @intent replace repeated suffix database scans with one transaction-local file-node snapshot. +type importIndexedLookup struct { + Store + fileNodesBySuffix map[string][]graph.Node + index *resolve.ImportFileIndex +} + +// newImportIndexedLookup creates a lookup decorator for a single staged edge-resolution phase. +// @intent scope cached import paths to one update transaction and avoid stale store-wide state. +func newImportIndexedLookup(store Store) *importIndexedLookup { + return &importIndexedLookup{ + Store: store, + fileNodesBySuffix: make(map[string][]graph.Node), + } +} + +// GetFileNodesByPathSuffix returns cached import candidates, initializing an immutable index once when supported. +// @intent preserve the legacy lookup fallback while avoiding repeated scans for staged bulk updates. +func (l *importIndexedLookup) GetFileNodesByPathSuffix(ctx context.Context, suffix string) ([]graph.Node, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + if nodes, ok := l.fileNodesBySuffix[suffix]; ok { + return nodes, nil + } + if l.index == nil { + lister, ok := l.Store.(importFileNodeLister) + if !ok { + legacyLookup, ok := l.Store.(fileSuffixLookup) + if !ok { + return nil, nil + } + return legacyLookup.GetFileNodesByPathSuffix(ctx, suffix) + } + nodes, err := lister.ListImportFileNodes(ctx) + if err != nil { + return nil, err + } + l.index = resolve.NewImportFileIndex(nodes) + } + nodes := l.index.Find(suffix) + l.fileNodesBySuffix[suffix] = nodes + return nodes, nil +} + +// GetEdgesToNodes forwards existing-edge reads when the underlying store supports them. +// @intent retain historical implements resolution while the lookup decorates import lookups. +func (l *importIndexedLookup) GetEdgesToNodes(ctx context.Context, nodeIDs []uint) ([]graph.Edge, error) { + reader, ok := l.Store.(edgeReader) + if !ok { + return nil, nil + } + return reader.GetEdgesToNodes(ctx, nodeIDs) +} diff --git a/internal/app/ingest/incremental/incremental.go b/internal/app/ingest/incremental/incremental.go index 792c79e..c1c8cd5 100644 --- a/internal/app/ingest/incremental/incremental.go +++ b/internal/app/ingest/incremental/incremental.go @@ -8,6 +8,8 @@ import ( "strconv" "strings" + "github.com/tae2089/trace" + "github.com/tae2089/code-context-graph/internal/app/ingest" "github.com/tae2089/code-context-graph/internal/app/ingest/binding" "github.com/tae2089/code-context-graph/internal/app/ingest/resolve" @@ -143,6 +145,25 @@ func (s *Syncer) SyncWithExistingStore(ctx context.Context, syncStore ingest.Gra return s.syncWithExisting(ctx, target, files, existingFiles) } +// SyncBatchesWithExisting stages every source batch before resolving any parsed edges. +// @intent prevent spool-record ordering from removing edges whose endpoints are both replaced in one bulk update. +// @param source bounded current-source batch replay owned by the workflow layer +// @param deletedFiles persisted paths absent from source that must be removed before edge replay +// @ensures edge resolution observes all current nodes after source staging completes +func (s *Syncer) SyncBatchesWithExisting(ctx context.Context, source ingest.FileBatchSource, deletedFiles []string) (*SyncStats, error) { + return s.syncBatchesWithExisting(ctx, s.store, source, deletedFiles) +} + +// SyncBatchesWithExistingStore stages reconciliation using the active transaction-scoped graph store. +// @intent keep bulk update node, edge, package, and search writes within one transaction. +func (s *Syncer) SyncBatchesWithExistingStore(ctx context.Context, syncStore ingest.GraphStore, source ingest.FileBatchSource, deletedFiles []string) (*SyncStats, error) { + var target Store = syncStore + if syncStore == nil { + target = s.store + } + return s.syncBatchesWithExisting(ctx, target, source, deletedFiles) +} + // syncWithExisting performs the actual diff-and-apply pass against the supplied store. // @intent compare hashes for known files, parse new/changed ones, and remove deleted entries in one pass. // @sideEffect upserts nodes/edges/annotations and deletes removed files through syncStore. @@ -221,7 +242,7 @@ func (s *Syncer) syncWithExisting(ctx context.Context, syncStore Store, files ma } } } - if err := s.resolveAndUpsertEdges(ctx, syncStore, parsedFiles, stats); err != nil { + if err := s.resolveAndUpsertEdges(ctx, syncStore, syncStore, parsedFiles, stats); err != nil { return nil, err } for _, parsed := range parsedFiles { @@ -253,14 +274,154 @@ func (s *Syncer) syncWithExisting(ctx context.Context, syncStore Store, files ma return stats, nil } +// syncBatchesWithExisting performs bounded node staging, deletion, then deferred edge replay. +// @intent make cross-file edge resolution independent of source batch ordering without retaining all parsed edges in memory. +// @sideEffect mutates nodes, annotations, edges, and temporary local spool files through syncStore. +// @domainRule no edge is resolved until every supplied source batch and deletion has completed. +func (s *Syncer) syncBatchesWithExisting(ctx context.Context, syncStore Store, source ingest.FileBatchSource, deletedFiles []string) (*SyncStats, error) { + if source == nil { + return nil, trace.New("incremental batch source is required") + } + + stats := &SyncStats{} + edgeSpool, err := newDeferredEdgeSpool() + if err != nil { + return nil, err + } + defer edgeSpool.cleanup(s.logger) + + s.logger.Info("staged sync started", "deleted_count", len(deletedFiles)) + if err := source(func(files map[string]FileInfo) error { + return s.stageBatch(ctx, syncStore, files, edgeSpool, stats) + }); err != nil { + return nil, err + } + + for _, filePath := range deletedFiles { + if err := syncStore.DeleteNodesByFile(ctx, filePath); err != nil { + return nil, err + } + s.logger.Debug("file deleted", "file", filePath) + stats.Deleted++ + } + + lookup := newImportIndexedLookup(syncStore) + for _, path := range edgeSpool.records { + record, err := edgeSpool.readRecord(path) + if err != nil { + return nil, err + } + parsedFiles := make([]parsedSyncFile, 0, len(record.Files)) + for _, file := range record.Files { + parsedFiles = append(parsedFiles, parsedSyncFile{filePath: file.FilePath, edges: file.Edges}) + } + if err := s.resolveAndUpsertEdges(ctx, syncStore, lookup, parsedFiles, stats); err != nil { + return nil, err + } + } + + s.logger.Info("staged sync completed", + "added", stats.Added, + "modified", stats.Modified, + "skipped", stats.Skipped, + "deleted", stats.Deleted, + "unresolved_edges", stats.Unresolved.DroppedCount, + "unresolved_by_kind", formatEdgeKindCounts(stats.Unresolved.ByKind), + "unresolved_by_file", stats.Unresolved.ByFile, + "unresolved_by_reason", stats.Unresolved.ByReason, + "unresolved_samples", stats.Unresolved.Samples, + ) + return stats, nil +} + +// stageBatch applies one bounded source batch and spools its parsed edges for the later resolution phase. +// @intent release source content after node and annotation writes while preserving only edges required for cross-file resolution. +// @sideEffect deletes and upserts graph nodes and annotations, then writes an invocation-local edge spool record. +func (s *Syncer) stageBatch(ctx context.Context, syncStore Store, files map[string]FileInfo, edgeSpool *deferredEdgeSpool, stats *SyncStats) error { + filePaths := make([]string, 0, len(files)) + for filePath := range files { + filePaths = append(filePaths, filePath) + } + existingByFile, err := syncStore.GetNodesByFiles(ctx, filePaths) + if err != nil { + return err + } + + orderedPaths := sortedFilePaths(files) + parsedFiles := make([]parsedSyncFile, 0, len(files)) + for _, filePath := range orderedPaths { + info := files[filePath] + existing := existingByFile[filePath] + parser := s.resolveParser(filePath) + if parser == nil { + s.logger.Debug("file skipped (no parser)", "file", filePath) + stats.Skipped++ + releaseContent(files, filePath) + continue + } + if len(existing) > 0 && existing[0].Hash == info.Hash && !info.Force { + s.logger.Debug("file skipped (unchanged)", "file", filePath) + stats.Skipped++ + releaseContent(files, filePath) + continue + } + + parsed := parsedSyncFile{filePath: filePath, info: info, hadExisting: len(existing) > 0} + if annotatingParser, ok := parser.(AnnotatingParser); ok { + parsed.nodes, parsed.edges, parsed.comments, err = annotatingParser.ParseWithComments(ctx, filePath, info.Content) + parsed.language = annotatingParser.Language() + } else { + parsed.nodes, parsed.edges, err = parser.Parse(filePath, info.Content) + } + if err != nil { + return err + } + setNodeHashes(parsed.nodes, info.Hash) + parsedFiles = append(parsedFiles, parsed) + } + + for _, parsed := range parsedFiles { + if !parsed.hadExisting { + s.logger.Debug("file added", "file", parsed.filePath) + stats.Added++ + continue + } + if err := syncStore.DeleteNodesByFile(ctx, parsed.filePath); err != nil { + return err + } + s.logger.Debug("file modified", "file", parsed.filePath) + stats.Modified++ + } + + edgeRecord := deferredEdgeRecord{Files: make([]deferredEdgeFile, 0, len(parsedFiles))} + for i := range parsedFiles { + parsed := &parsedFiles[i] + if len(parsed.nodes) > 0 { + if err := syncStore.UpsertNodes(ctx, parsed.nodes); err != nil { + return err + } + if len(parsed.comments) > 0 { + if err := s.restoreAnnotations(ctx, syncStore, parsed.filePath, parsed.info.Content, parsed.nodes, parsed.comments, parsed.language); err != nil { + return err + } + } + } + if len(parsed.edges) > 0 { + edgeRecord.Files = append(edgeRecord.Files, deferredEdgeFile{FilePath: parsed.filePath, Edges: parsed.edges}) + } + releaseContent(files, parsed.filePath) + } + return edgeSpool.writeRecord(edgeRecord) +} + // resolveAndUpsertEdges resolves parsed edges in dependency-safe phases before persisting them. // @intent preserve interface dispatch and import-backed call resolution during incremental sync updates. // @sideEffect upserts resolved graph edges through the sync store. // @mutates graph edges, stats.Unresolved -func (s *Syncer) resolveAndUpsertEdges(ctx context.Context, syncStore Store, parsedFiles []parsedSyncFile, stats *SyncStats) error { +func (s *Syncer) resolveAndUpsertEdges(ctx context.Context, syncStore Store, lookup resolve.NodeLookup, parsedFiles []parsedSyncFile, stats *SyncStats) error { implementsEdges, otherByFile := partitionParsedSyncEdges(parsedFiles) for _, edgeChunk := range splitEdgeChunks(implementsEdges) { - resolved, err := resolve.ResolveWithOptions(ctx, syncStore, edgeChunk, s.opts) + resolved, err := resolve.ResolveWithOptions(ctx, lookup, edgeChunk, s.opts) if err != nil { return err } @@ -278,7 +439,7 @@ func (s *Syncer) resolveAndUpsertEdges(ctx context.Context, syncStore Store, par edges := otherByFile[parsed.filePath] for _, edgeChunk := range splitEdgeChunks(edges) { resolveInput := chunkWithImportWarmup(edgeChunk, importsByFile[parsed.filePath]) - resolved, err := resolve.ResolveWithOptions(ctx, syncStore, resolveInput, s.opts) + resolved, err := resolve.ResolveWithOptions(ctx, lookup, resolveInput, s.opts) if err != nil { return err } diff --git a/internal/app/ingest/ports.go b/internal/app/ingest/ports.go index 45f7e72..8f4dfc5 100644 --- a/internal/app/ingest/ports.go +++ b/internal/app/ingest/ports.go @@ -168,3 +168,24 @@ type IncrementalSyncer interface { type TransactionalIncrementalSyncer interface { SyncWithExistingStore(ctx context.Context, graphStore GraphStore, files map[string]FileInfo, existingFiles []string) (*SyncStats, error) } + +// FileBatchVisitor receives one bounded source batch during staged incremental reconciliation. +// @intent keep bulk update input streaming while allowing a syncer to own cross-batch ordering. +type FileBatchVisitor func(files map[string]FileInfo) error + +// FileBatchSource replays the current source snapshot in bounded batches. +// @intent let workflow retain source spooling while incremental reconciliation controls node and edge phases. +type FileBatchSource func(visitor FileBatchVisitor) error + +// BatchIncrementalSyncer supports a single staged reconciliation across multiple input batches. +// @intent prevent batch order from affecting cross-file edge resolution during large updates. +// @domainRule deletedFiles must contain only paths absent from the supplied source batches. +type BatchIncrementalSyncer interface { + SyncBatchesWithExisting(ctx context.Context, source FileBatchSource, deletedFiles []string) (*SyncStats, error) +} + +// TransactionalBatchIncrementalSyncer runs staged reconciliation against the active graph transaction. +// @intent preserve one atomic graph and search transaction while reconciling streamed update batches. +type TransactionalBatchIncrementalSyncer interface { + SyncBatchesWithExistingStore(ctx context.Context, graphStore GraphStore, source FileBatchSource, deletedFiles []string) (*SyncStats, error) +} diff --git a/internal/app/ingest/resolve/import_file_index.go b/internal/app/ingest/resolve/import_file_index.go new file mode 100644 index 0000000..bc0ea30 --- /dev/null +++ b/internal/app/ingest/resolve/import_file_index.go @@ -0,0 +1,63 @@ +// @index Immutable exact and longest-suffix file-node index shared by build and update resolution. +package resolve + +import ( + "path" + "strings" + + "github.com/tae2089/code-context-graph/internal/domain/graph" +) + +// ImportFileIndex maps exact directories and their suffixes to persisted file nodes. +// @intent resolve many import paths from one immutable file-node snapshot without repeated store scans. +type ImportFileIndex struct { + byDirectory map[string][]graph.Node + bySuffix map[string][]graph.Node +} + +// NewImportFileIndex precomputes directory suffixes for the supplied file nodes. +// @intent share the exact-directory and longest-suffix import policy across build and staged update resolution. +func NewImportFileIndex(nodes []graph.Node) *ImportFileIndex { + index := &ImportFileIndex{ + byDirectory: make(map[string][]graph.Node), + bySuffix: make(map[string][]graph.Node), + } + for _, node := range nodes { + if node.Kind != graph.NodeKindFile { + continue + } + dir := strings.Trim(path.Dir(node.FilePath), "/") + if dir == "" || dir == "." { + continue + } + index.byDirectory[dir] = append(index.byDirectory[dir], node) + parts := strings.Split(dir, "/") + for start := range parts { + suffix := strings.Join(parts[start:], "/") + index.bySuffix[suffix] = append(index.bySuffix[suffix], node) + } + } + return index +} + +// Find returns exact directory matches first, then all matches with the longest suffix. +// @intent preserve GraphStore import lookup precedence using bounded map reads. +func (i *ImportFileIndex) Find(importPath string) []graph.Node { + if i == nil { + return nil + } + importPath = strings.Trim(path.Clean(strings.TrimSpace(importPath)), "/") + if importPath == "" || importPath == "." { + return nil + } + if exact := i.byDirectory[importPath]; len(exact) > 0 { + return exact + } + parts := strings.Split(importPath, "/") + for start := range parts { + if candidates := i.bySuffix[strings.Join(parts[start:], "/")]; len(candidates) > 0 { + return candidates + } + } + return nil +} diff --git a/internal/app/ingest/workflow/build.go b/internal/app/ingest/workflow/build.go index 7433d40..7e7a49f 100644 --- a/internal/app/ingest/workflow/build.go +++ b/internal/app/ingest/workflow/build.go @@ -7,7 +7,6 @@ import ( "encoding/hex" "errors" "os" - "path" "path/filepath" "strconv" "strings" @@ -205,6 +204,13 @@ func (s *Service) Build(ctx context.Context, opts BuildOptions) (BuildStats, err "files", stats.TotalFiles, "nodes", stats.TotalNodes, "edges", stats.TotalEdges, + "unresolved_edges", stats.Unresolved.DroppedCount, + "unresolved_by_kind", formatEdgeKindCounts(stats.Unresolved.ByKind), + "unresolved_by_file", stats.Unresolved.ByFile, + "unresolved_by_reason", stats.Unresolved.ByReason, + "unresolved_samples", stats.Unresolved.Samples, + ) + s.logger().Debug("build timing", "parse_ms", stats.Timing.ParseMS, "persist_nodes_ms", stats.Timing.PersistNodesMS, "resolve_edges_ms", stats.Timing.ResolveEdgesMS, @@ -224,11 +230,6 @@ func (s *Service) Build(ctx context.Context, opts BuildOptions) (BuildStats, err "resolve_edge_upsert_ms", stats.Timing.Resolve.UpsertEdges.MS, "search_rebuild_ms", stats.Timing.SearchRebuildMS, "total_ms", stats.Timing.TotalMS, - "unresolved_edges", stats.Unresolved.DroppedCount, - "unresolved_by_kind", formatEdgeKindCounts(stats.Unresolved.ByKind), - "unresolved_by_file", stats.Unresolved.ByFile, - "unresolved_by_reason", stats.Unresolved.ByReason, - "unresolved_samples", stats.Unresolved.Samples, ) return stats, nil @@ -627,64 +628,10 @@ func (s *Service) flushBuildBatch(ctx context.Context, txStore ingestapp.GraphSt type buildResolveLookup struct { ingestapp.GraphStore fileNodesBySuffix map[string][]graph.Node - importFileIndex *buildImportFileIndex + importFileIndex *resolve.ImportFileIndex timing *BuildResolveTiming } -// buildImportFileIndex maps exact directories and their suffixes to persisted file nodes. -// @intent resolve many imports from one immutable build-phase file-node snapshot. -type buildImportFileIndex struct { - byDirectory map[string][]graph.Node - bySuffix map[string][]graph.Node -} - -// newBuildImportFileIndex precomputes directory suffixes for the current build's file nodes. -// @intent turn repeated longest-suffix scans into bounded map lookups during edge resolution. -func newBuildImportFileIndex(nodes []graph.Node) *buildImportFileIndex { - index := &buildImportFileIndex{ - byDirectory: make(map[string][]graph.Node), - bySuffix: make(map[string][]graph.Node), - } - for _, node := range nodes { - if node.Kind != graph.NodeKindFile { - continue - } - dir := strings.Trim(path.Dir(node.FilePath), "/") - if dir == "" || dir == "." { - continue - } - index.byDirectory[dir] = append(index.byDirectory[dir], node) - parts := strings.Split(dir, "/") - for start := range parts { - suffix := strings.Join(parts[start:], "/") - index.bySuffix[suffix] = append(index.bySuffix[suffix], node) - } - } - return index -} - -// find returns exact directory matches first, then all longest suffix matches. -// @intent preserve GraphStore import resolution precedence while avoiding repeated database scans. -func (i *buildImportFileIndex) find(importPath string) []graph.Node { - if i == nil { - return nil - } - importPath = strings.Trim(path.Clean(strings.TrimSpace(importPath)), "/") - if importPath == "" || importPath == "." { - return nil - } - if exact := i.byDirectory[importPath]; len(exact) > 0 { - return exact - } - parts := strings.Split(importPath, "/") - for start := range parts { - if candidates := i.bySuffix[strings.Join(parts[start:], "/")]; len(candidates) > 0 { - return candidates - } - } - return nil -} - // newBuildResolveLookup creates the build-scoped read-through lookup cache. // @intent share immutable import file-node results across all resolver chunks in one build. func newBuildResolveLookup(store ingestapp.GraphStore) *buildResolveLookup { @@ -773,9 +720,9 @@ func (l *buildResolveLookup) GetFileNodesByPathSuffix(ctx context.Context, suffi if err != nil { return nil, err } - l.importFileIndex = newBuildImportFileIndex(nodes) + l.importFileIndex = resolve.NewImportFileIndex(nodes) } - nodes := l.importFileIndex.find(suffix) + nodes := l.importFileIndex.Find(suffix) l.fileNodesBySuffix[suffix] = nodes return nodes, nil } diff --git a/internal/app/ingest/workflow/indexer.go b/internal/app/ingest/workflow/indexer.go index 13cbd78..1334809 100644 --- a/internal/app/ingest/workflow/indexer.go +++ b/internal/app/ingest/workflow/indexer.go @@ -39,6 +39,12 @@ type IncrementalSyncer = ingestapp.IncrementalSyncer // @intent allow incremental sync implementations to participate in the same transaction-scoped store used by Service updates. type transactionalIncrementalSyncer = ingestapp.TransactionalIncrementalSyncer +// @intent allow incremental implementations to reconcile all update spool batches before resolving cross-file edges. +type batchIncrementalSyncer = ingestapp.BatchIncrementalSyncer + +// @intent let staged incremental reconciliation use the graph store from the active update transaction. +type transactionalBatchIncrementalSyncer = ingestapp.TransactionalBatchIncrementalSyncer + // Service orchestrates graph building and search document refresh. // @intent 파싱 결과 저장과 검색 인덱스 재구성을 하나의 서비스로 묶는다. type Service struct { diff --git a/internal/app/ingest/workflow/indexer_test.go b/internal/app/ingest/workflow/indexer_test.go index e477711..85c6017 100644 --- a/internal/app/ingest/workflow/indexer_test.go +++ b/internal/app/ingest/workflow/indexer_test.go @@ -10,6 +10,7 @@ package workflow import ( + "bytes" "context" "crypto/sha256" "encoding/hex" @@ -905,6 +906,129 @@ func TestUpdate_EmitsCrossFileGoStructuralImplements(t *testing.T) { t.Fatalf("expected update-time cross-file implements edge from %d to %d, got %+v", impl.ID, iface.ID, edges) } +func TestUpdate_ReconcilesCrossBatchChangedCallEdge(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{Logger: gormlogger.Discard}) + if err != nil { + t.Fatalf("open db: %v", err) + } + st := graphgorm.New(db) + if err := st.AutoMigrate(); err != nil { + t.Fatalf("migrate: %v", err) + } + walker := treesitter.NewWalker(treesitter.GoSpec) + svc := &Service{Store: st, UnitOfWork: newTestUnitOfWork(db, nil), Walkers: map[string]Parser{".go": walker}, Logger: slog.Default()} + + tmpDir := t.TempDir() + mustWrite := func(rel, content string) { + full := filepath.Join(tmpDir, rel) + if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil { + t.Fatalf("mkdir %s: %v", rel, err) + } + if err := os.WriteFile(full, []byte(content), 0o644); err != nil { + t.Fatalf("write %s: %v", rel, err) + } + } + mustWrite("a_source.go", "package sample\n\nfunc Source() { Target() }\n") + for i := range buildFlushFileBatchSize - 1 { + mustWrite(fmt.Sprintf("m_filler_%03d.go", i), fmt.Sprintf("package sample\n\nfunc Filler%03d() {}\n", i)) + } + mustWrite("z_target.go", "package sample\n\nfunc Target() {}\n") + + ctx := context.Background() + if _, err := svc.Build(ctx, BuildOptions{Dir: tmpDir}); err != nil { + t.Fatalf("Build: %v", err) + } + + mustWrite("a_source.go", "package sample\n\nfunc Source() { Target() }\n\n// source changed\n") + mustWrite("z_target.go", "package sample\n\nfunc Target() {}\n\n// target changed\n") + syncer := incremental.NewWithRegistry(st, map[string]incremental.Parser{".go": walker}, incremental.WithLogger(slog.Default())) + stats, err := svc.Update(ctx, UpdateOptions{BuildOptions: BuildOptions{Dir: tmpDir}, Syncer: syncer}) + if err != nil { + t.Fatalf("Update: %v", err) + } + if stats.Modified != 2 { + t.Fatalf("Modified = %d, want 2", stats.Modified) + } + + source, err := st.GetNode(ctx, "sample.Source") + if err != nil || source == nil { + t.Fatalf("GetNode source: node=%v err=%v", source, err) + } + target, err := st.GetNode(ctx, "sample.Target") + if err != nil || target == nil { + t.Fatalf("GetNode target: node=%v err=%v", target, err) + } + edges, err := st.GetEdgesFrom(ctx, source.ID) + if err != nil { + t.Fatalf("GetEdgesFrom: %v", err) + } + for _, edge := range edges { + if edge.Kind == graph.EdgeKindCalls && edge.ToNodeID == target.ID { + return + } + } + t.Fatalf("expected cross-batch call edge from %d to %d, got %+v", source.ID, target.ID, edges) +} + +func TestUpdate_ReconcilesExistingCallerAfterTargetFileAdded(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{Logger: gormlogger.Discard}) + if err != nil { + t.Fatalf("open db: %v", err) + } + st := graphgorm.New(db) + if err := st.AutoMigrate(); err != nil { + t.Fatalf("migrate: %v", err) + } + walker := treesitter.NewWalker(treesitter.GoSpec) + svc := &Service{Store: st, UnitOfWork: newTestUnitOfWork(db, nil), Walkers: map[string]Parser{".go": walker}, Logger: slog.Default()} + + tmpDir := t.TempDir() + writeFile := func(rel, content string) { + full := filepath.Join(tmpDir, rel) + if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil { + t.Fatalf("mkdir %s: %v", rel, err) + } + if err := os.WriteFile(full, []byte(content), 0o644); err != nil { + t.Fatalf("write %s: %v", rel, err) + } + } + writeFile("source.go", "package sample\n\nfunc Source() { Target() }\n") + + ctx := context.Background() + if _, err := svc.Build(ctx, BuildOptions{Dir: tmpDir}); err != nil { + t.Fatalf("Build before target exists: %v", err) + } + writeFile("target.go", "package sample\n\nfunc Target() {}\n") + + syncer := incremental.NewWithRegistry(st, map[string]incremental.Parser{".go": walker}, incremental.WithLogger(slog.Default())) + stats, err := svc.Update(ctx, UpdateOptions{BuildOptions: BuildOptions{Dir: tmpDir}, Syncer: syncer}) + if err != nil { + t.Fatalf("Update after target add: %v", err) + } + if stats.Added != 1 { + t.Fatalf("Added = %d, want one newly added target", stats.Added) + } + + source, err := st.GetNode(ctx, "sample.Source") + if err != nil || source == nil { + t.Fatalf("GetNode source: node=%v err=%v", source, err) + } + target, err := st.GetNode(ctx, "sample.Target") + if err != nil || target == nil { + t.Fatalf("GetNode target: node=%v err=%v", target, err) + } + edges, err := st.GetEdgesFrom(ctx, source.ID) + if err != nil { + t.Fatalf("GetEdgesFrom: %v", err) + } + for _, edge := range edges { + if edge.Kind == graph.EdgeKindCalls && edge.ToNodeID == target.ID { + return + } + } + t.Fatalf("expected existing caller edge from %d to newly added target %d, got %+v", source.ID, target.ID, edges) +} + func TestUpdate_RemovesStaleCrossFileGoStructuralImplements(t *testing.T) { db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{Logger: gormlogger.Discard}) if err != nil { @@ -2304,6 +2428,80 @@ func Keep` + strconv.Itoa(i) + `() {} } } +func TestBuildCompletionLogsTimingOnlyAtDebugLevel(t *testing.T) { + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "source.go"), []byte("package sample\n\nfunc Source() {}\n"), 0o644); err != nil { + t.Fatalf("write source: %v", err) + } + newService := func(logger *slog.Logger) *Service { + store := newRecordingGraphStore(t) + return &Service{ + Store: store, + UnitOfWork: testUnitOfWork{graph: store}, + Walkers: map[string]Parser{ + ".go": treesitter.NewWalker(treesitter.GoSpec), + }, + Logger: logger, + } + } + timingKeys := []string{ + "parse_ms=", + "persist_nodes_ms=", + "resolve_edges_ms=", + "resolver_calls=", + "resolver_ms=", + "resolve_nodes_by_ids_calls=", + "resolve_nodes_by_ids_ms=", + "resolve_nodes_by_files_calls=", + "resolve_nodes_by_files_ms=", + "resolve_nodes_by_qn_calls=", + "resolve_nodes_by_qn_ms=", + "resolve_import_file_nodes_calls=", + "resolve_import_file_nodes_ms=", + "resolve_edges_to_nodes_calls=", + "resolve_edges_to_nodes_ms=", + "resolve_edge_upsert_calls=", + "resolve_edge_upsert_ms=", + "search_rebuild_ms=", + "total_ms=", + } + + var infoLogs bytes.Buffer + if _, err := newService(slog.New(slog.NewTextHandler(&infoLogs, &slog.HandlerOptions{Level: slog.LevelInfo}))).Build(context.Background(), BuildOptions{Dir: dir}); err != nil { + t.Fatalf("info Build: %v", err) + } + infoOutput := infoLogs.String() + if !strings.Contains(infoOutput, `msg="build complete"`) || !strings.Contains(infoOutput, "files=1") { + t.Fatalf("expected info completion summary, got %q", infoOutput) + } + if strings.Contains(infoOutput, `msg="build timing"`) { + t.Fatalf("info output must not contain debug timing event, got %q", infoOutput) + } + for _, key := range timingKeys { + if strings.Contains(infoOutput, key) { + t.Fatalf("info completion log must omit %q, got %q", key, infoOutput) + } + } + + var debugLogs bytes.Buffer + stats, err := newService(slog.New(slog.NewTextHandler(&debugLogs, &slog.HandlerOptions{Level: slog.LevelDebug}))).Build(context.Background(), BuildOptions{Dir: dir}) + if err != nil { + t.Fatalf("debug Build: %v", err) + } + if stats.Timing.TotalMS < 0 { + t.Fatalf("Timing.TotalMS = %d, want non-negative", stats.Timing.TotalMS) + } + debugOutput := debugLogs.String() + if !strings.Contains(debugOutput, `msg="build timing"`) { + t.Fatalf("expected debug timing event, got %q", debugOutput) + } + for _, key := range timingKeys { + if !strings.Contains(debugOutput, key) { + t.Fatalf("debug timing log must contain %q, got %q", key, debugOutput) + } + } +} + func TestBuild_ReleasesBatchCommentStateAfterBinding(t *testing.T) { var snapshots []struct { batch int diff --git a/internal/app/ingest/workflow/packages.go b/internal/app/ingest/workflow/packages.go index 25a8a0f..985f9d3 100644 --- a/internal/app/ingest/workflow/packages.go +++ b/internal/app/ingest/workflow/packages.go @@ -504,6 +504,47 @@ func affectedPackageImportPaths(packages map[string]languagePackageInfo, affecte return affected } +// addUnchangedPeersForAddedFiles marks unchanged persisted peers for reparse when a new file joins their package or directory. +// @intent let local symbol lookup observe declarations introduced by newly added source files during incremental update. +// @domainRule discovered packages cover multi-directory language packages; a same-directory fallback preserves local resolution when discovery is unavailable. +// @mutates forceFiles +func addUnchangedPeersForAddedFiles(forceFiles map[string]struct{}, packages map[string]languagePackageInfo, existingNodesByFile map[string][]graph.Node, currentHashes map[string]string) { + if forceFiles == nil || len(existingNodesByFile) == 0 || len(currentHashes) == 0 { + return + } + addedFiles := make([]string, 0) + addedDirs := make(map[string]struct{}) + for filePath := range currentHashes { + if len(existingNodesByFile[filePath]) == 0 { + addedFiles = append(addedFiles, filePath) + addedDirs[path.Dir(filepath.ToSlash(filePath))] = struct{}{} + } + } + if len(addedFiles) == 0 { + return + } + peerFiles := make(map[string]struct{}) + for _, importPath := range affectedPackageImportPaths(packages, addedFiles) { + for _, filePath := range packages[importPath].Files { + peerFiles[filepath.ToSlash(filePath)] = struct{}{} + } + } + for filePath := range existingNodesByFile { + filePath = filepath.ToSlash(filePath) + if _, sameDir := addedDirs[path.Dir(filePath)]; sameDir { + peerFiles[filePath] = struct{}{} + } + } + for filePath := range peerFiles { + nodes := existingNodesByFile[filePath] + currentHash, present := currentHashes[filePath] + if len(nodes) == 0 || !present || nodes[0].Hash != currentHash { + continue + } + forceFiles[filePath] = struct{}{} + } +} + // appendUniqueString appends a string to a slice only if it is not already present. // @intent maintain a unique set of strings while preserving insertion order for small sets. func appendUniqueString(values []string, value string) []string { diff --git a/internal/app/ingest/workflow/update.go b/internal/app/ingest/workflow/update.go index 85d529f..1b97e14 100644 --- a/internal/app/ingest/workflow/update.go +++ b/internal/app/ingest/workflow/update.go @@ -174,7 +174,7 @@ func (s *Service) prepareUpdateSpool(ctx context.Context, absDir string, opts Up } // applyUpdateSpoolInTx replays the update spool through the incremental syncer and refreshes affected search docs. -// @intent stage normal-changed files first, then deletions, then forced reparses so edge-source files always observe up-to-date nodes. +// @intent prefer staged reconciliation so every changed or forced file node is current before any cross-file edge is resolved. // @sideEffect adds, modifies, and deletes graph nodes/edges/annotations for changed files and refreshes affected search documents. // @mutates graph nodes, edges, annotations, and search_documents func (s *Service) applyUpdateSpoolInTx(ctx context.Context, tx ingest.Transaction, opts UpdateOptions, spool *updateSpool) (*ingest.SyncStats, error) { @@ -199,52 +199,58 @@ func (s *Service) applyUpdateSpoolInTx(ctx context.Context, tx ingest.Transactio if err != nil { return nil, trace.Wrap(err, "load edge source files for changed graph") } + addUnchangedPeersForAddedFiles(forceFiles, spool.packages, existingNodesByFile, spool.currentHashes) spool.forceFiles = forceFiles - - for _, path := range spool.records { - record, err := spool.readRecord(path) - if err != nil { - return nil, err - } - normalFiles, _ := splitForcedFiles(record.Files, spool.forceFiles) - if len(normalFiles) == 0 { - continue - } - batchStats, err := syncIncrementalBatch(ctx, syncer, txStore, normalFiles, nil) - if err != nil { - return nil, trace.Wrap(err, "incremental sync") - } - addSyncStats(stats, batchStats) - } - deletedFiles := make([]string, 0, len(existingFiles)) for _, fp := range existingFiles { if _, ok := spool.currentFiles[fp]; !ok { deletedFiles = append(deletedFiles, fp) } } - if len(deletedFiles) > 0 { - batchStats, err := syncIncrementalBatch(ctx, syncer, txStore, nil, deletedFiles) + if stagedSyncer, ok := syncer.(transactionalBatchIncrementalSyncer); ok { + stats, err = stagedSyncer.SyncBatchesWithExistingStore(ctx, txStore, newUpdateSpoolBatchSource(ctx, spool), deletedFiles) if err != nil { - return nil, trace.Wrap(err, "incremental sync") + return nil, trace.Wrap(err, "staged incremental sync") } - addSyncStats(stats, batchStats) - } - - for _, path := range spool.records { - record, err := spool.readRecord(path) - if err != nil { - return nil, err + } else { + for _, path := range spool.records { + record, err := spool.readRecord(path) + if err != nil { + return nil, err + } + normalFiles, _ := splitForcedFiles(record.Files, spool.forceFiles) + if len(normalFiles) == 0 { + continue + } + batchStats, err := syncIncrementalBatch(ctx, syncer, txStore, normalFiles, nil) + if err != nil { + return nil, trace.Wrap(err, "incremental sync") + } + addSyncStats(stats, batchStats) } - _, forcedFiles := splitForcedFiles(record.Files, spool.forceFiles) - if len(forcedFiles) == 0 { - continue + if len(deletedFiles) > 0 { + batchStats, err := syncIncrementalBatch(ctx, syncer, txStore, nil, deletedFiles) + if err != nil { + return nil, trace.Wrap(err, "incremental sync") + } + addSyncStats(stats, batchStats) } - batchStats, err := syncIncrementalBatch(ctx, syncer, txStore, forcedFiles, nil) - if err != nil { - return nil, trace.Wrap(err, "incremental force sync") + + for _, path := range spool.records { + record, err := spool.readRecord(path) + if err != nil { + return nil, err + } + _, forcedFiles := splitForcedFiles(record.Files, spool.forceFiles) + if len(forcedFiles) == 0 { + continue + } + batchStats, err := syncIncrementalBatch(ctx, syncer, txStore, forcedFiles, nil) + if err != nil { + return nil, trace.Wrap(err, "incremental force sync") + } + addSyncStats(stats, batchStats) } - addSyncStats(stats, batchStats) } changedFiles := affectedUpdateFiles(spool.currentHashes, existingNodesByFile, spool.forceFiles) if err := s.refreshPackageSemanticEdges(ctx, txStore, opts.Dir, spool.packages, changedFiles, deletedFiles, resolve.ResolveOptions{FallbackCalls: opts.FallbackCalls}); err != nil { @@ -279,6 +285,7 @@ func (s *Service) updateGraphWithoutTx(ctx context.Context, absDir string, opts if err != nil { return nil, trace.Wrap(err, "load edge source files for changed graph") } + addUnchangedPeersForAddedFiles(forceFiles, packages, existingNodesByFile, spool.currentHashes) spool.forceFiles = forceFiles if s.Store != nil { if err := upsertPackageNodes(ctx, s.Store, packages); err != nil { @@ -286,48 +293,55 @@ func (s *Service) updateGraphWithoutTx(ctx context.Context, absDir string, opts } } - stats := &ingest.SyncStats{} - // Normal batches must never carry existingFiles: SyncWithExisting deletes existing files - // absent from the batch, so a multi-record spool would delete files belonging to later - // batches (then re-add them, churning node IDs and stats). Deletions are handled once, - // explicitly, below — mirroring the transactional path (applyUpdateSpoolInTx). - for _, path := range spool.records { - record, err := spool.readRecord(path) - if err != nil { - return nil, err - } - normalFiles, _ := splitForcedFiles(record.Files, spool.forceFiles) - if len(normalFiles) == 0 { - continue - } - batchStats, err := opts.Syncer.SyncWithExisting(ctx, normalFiles, nil) - if err != nil { - return nil, trace.Wrap(err, "incremental sync") - } - addSyncStats(stats, batchStats) - } deletedFiles := existingFilesMissingFromSet(spool.currentFiles, existingFiles) - if len(deletedFiles) > 0 { - batchStats, err := opts.Syncer.SyncWithExisting(ctx, nil, deletedFiles) - if err != nil { - return nil, trace.Wrap(err, "incremental delete sync") - } - addSyncStats(stats, batchStats) - } - for _, path := range spool.records { - record, err := spool.readRecord(path) + stats := &ingest.SyncStats{} + if stagedSyncer, ok := opts.Syncer.(batchIncrementalSyncer); ok { + stats, err = stagedSyncer.SyncBatchesWithExisting(ctx, newUpdateSpoolBatchSource(ctx, spool), deletedFiles) if err != nil { - return nil, err + return nil, trace.Wrap(err, "staged incremental sync") + } + } else { + // Normal batches must never carry existingFiles: SyncWithExisting deletes existing files + // absent from the batch, so a multi-record spool would delete files belonging to later + // batches (then re-add them, churning node IDs and stats). Deletions are handled once, + // explicitly, below — mirroring the transactional path (applyUpdateSpoolInTx). + for _, path := range spool.records { + record, err := spool.readRecord(path) + if err != nil { + return nil, err + } + normalFiles, _ := splitForcedFiles(record.Files, spool.forceFiles) + if len(normalFiles) == 0 { + continue + } + batchStats, err := opts.Syncer.SyncWithExisting(ctx, normalFiles, nil) + if err != nil { + return nil, trace.Wrap(err, "incremental sync") + } + addSyncStats(stats, batchStats) } - _, forcedFiles := splitForcedFiles(record.Files, spool.forceFiles) - if len(forcedFiles) == 0 { - continue + if len(deletedFiles) > 0 { + batchStats, err := opts.Syncer.SyncWithExisting(ctx, nil, deletedFiles) + if err != nil { + return nil, trace.Wrap(err, "incremental delete sync") + } + addSyncStats(stats, batchStats) } - batchStats, err := opts.Syncer.SyncWithExisting(ctx, forcedFiles, nil) - if err != nil { - return nil, trace.Wrap(err, "incremental force sync") + for _, path := range spool.records { + record, err := spool.readRecord(path) + if err != nil { + return nil, err + } + _, forcedFiles := splitForcedFiles(record.Files, spool.forceFiles) + if len(forcedFiles) == 0 { + continue + } + batchStats, err := opts.Syncer.SyncWithExisting(ctx, forcedFiles, nil) + if err != nil { + return nil, trace.Wrap(err, "incremental force sync") + } + addSyncStats(stats, batchStats) } - addSyncStats(stats, batchStats) } changedFiles := affectedUpdateFiles(spool.currentHashes, existingNodesByFile, spool.forceFiles) if err := s.refreshPackageSemanticEdges(ctx, s.Store, absDir, packages, changedFiles, deletedFiles, resolve.ResolveOptions{FallbackCalls: opts.FallbackCalls}); err != nil { @@ -352,6 +366,32 @@ func (s *Service) updateGraphWithoutTx(ctx context.Context, absDir string, opts return stats, nil } +// newUpdateSpoolBatchSource replays all current update inputs while marking forced reparses in their original batch. +// @intent let a staged syncer own cross-batch ordering without loading the entire source snapshot into memory. +func newUpdateSpoolBatchSource(ctx context.Context, spool *updateSpool) ingest.FileBatchSource { + return func(visitor ingest.FileBatchVisitor) error { + for _, path := range spool.records { + if err := ctx.Err(); err != nil { + return err + } + record, err := spool.readRecord(path) + if err != nil { + return err + } + for filePath, info := range record.Files { + if _, forced := spool.forceFiles[filePath]; forced { + info.Force = true + record.Files[filePath] = info + } + } + if err := visitor(record.Files); err != nil { + return err + } + } + return nil + } +} + // affectedUpdateFiles selects files whose stored hash differs from the current input or that are forced to reparse. // @intent identify which files contributed nodes that need to be re-indexed for search after an incremental update. func affectedUpdateFiles(currentHashes map[string]string, existingNodesByFile map[string][]graph.Node, forceFiles map[string]struct{}) []string { From 03c3963d3308f79efb2b67dc5f5658191ced6d70 Mon Sep 17 00:00:00 2001 From: tae2089 Date: Thu, 16 Jul 2026 09:54:54 +0900 Subject: [PATCH 2/2] perf: batch node persistence during builds --- internal/app/ingest/workflow/build.go | 56 ++++++++++---- internal/app/ingest/workflow/indexer_test.go | 80 +++++++++++++++++++- 2 files changed, 119 insertions(+), 17 deletions(-) diff --git a/internal/app/ingest/workflow/build.go b/internal/app/ingest/workflow/build.go index 7e7a49f..db2720e 100644 --- a/internal/app/ingest/workflow/build.go +++ b/internal/app/ingest/workflow/build.go @@ -107,26 +107,19 @@ func (b *buildPersistBatch) reset() { b.bytes = 0 } -// bindAndReleaseNodeBatch upserts a parsed file's nodes and binds its comment annotations within a transaction-scoped store. -// @intent persist nodes and their annotation bindings atomically per file before releasing comment buffers. -// @sideEffect writes graph nodes and annotation rows via the transaction-scoped store. -// @mutates graph nodes and annotations -func (s *Service) bindAndReleaseNodeBatch(ctx context.Context, txStore ingestapp.GraphStore, batches []parsedBuildNodeBatch, idx int) error { +// bindAndReleaseNodeBatch binds a parsed file's comments after its nodes have persisted, then releases comment buffers. +// @intent preserve per-file annotation binding and release behavior after the enclosing flush persists all nodes together. +// @sideEffect writes annotation rows via the transaction-scoped store. +// @mutates graph annotations +func (s *Service) bindAndReleaseNodeBatch(ctx context.Context, txStore ingestapp.GraphStore, storedNodesByFile map[string][]graph.Node, batches []parsedBuildNodeBatch, idx int) error { parsed := &batches[idx] - if err := txStore.UpsertNodes(ctx, parsed.nodes); err != nil { - return trace.Wrap(err, "upsert nodes for "+parsed.relPath) - } - if len(parsed.tsComments) > 0 { binderComments := toBinderComments(parsed.tsComments) binder := binding.NewBinder() bindings := binder.Bind(binderComments, parsed.nodes, parsed.language, parsed.sourceLines) - storedNodes, err := txStore.GetNodesByFile(ctx, parsed.relPath) - if err != nil { - return trace.Wrap(err, "get stored nodes for annotations") - } + storedNodes := storedNodesByFile[parsed.relPath] storedMap := make(map[string]*graph.Node, len(storedNodes)) for i := range storedNodes { key := storedNodes[i].QualifiedName + ":" + strconv.Itoa(storedNodes[i].StartLine) @@ -602,19 +595,50 @@ func (s *Service) packageSemanticEdgeBatches(batches []parsedBuildNodeBatch) []p return out } -// flushBuildBatch persists the buffered nodes for the current batch. -// @intent persist nodes before all edges so foreign-key style references can resolve. +// flushBuildBatch persists the buffered nodes for the current bounded batch. +// @intent persist all batch nodes before annotations and all edges so references can resolve with fewer store operations. // @sideEffect upserts graph nodes and annotations through the transaction-scoped store. // @mutates graph nodes and annotations func (s *Service) flushBuildBatch(ctx context.Context, txStore ingestapp.GraphStore, batch *buildPersistBatch) error { if batch.files == 0 { return nil } + if err := ctx.Err(); err != nil { + return err + } + + nodeCount := 0 + annotationFilePaths := make([]string, 0, len(batch.nodeBatches)) + for _, parsed := range batch.nodeBatches { + nodeCount += len(parsed.nodes) + if len(parsed.tsComments) > 0 { + annotationFilePaths = append(annotationFilePaths, parsed.relPath) + } + } + nodes := make([]graph.Node, 0, nodeCount) + for _, parsed := range batch.nodeBatches { + nodes = append(nodes, parsed.nodes...) + } + if len(nodes) > 0 { + if err := txStore.UpsertNodes(ctx, nodes); err != nil { + return trace.Wrap(err, "upsert batch nodes") + } + } + + var storedNodesByFile map[string][]graph.Node + if len(annotationFilePaths) > 0 { + stored, err := txStore.GetNodesByFiles(ctx, annotationFilePaths) + if err != nil { + return trace.Wrap(err, "get stored nodes for annotations") + } + storedNodesByFile = stored + } + for i := range batch.nodeBatches { if err := ctx.Err(); err != nil { return err } - if err := s.bindAndReleaseNodeBatch(ctx, txStore, batch.nodeBatches, i); err != nil { + if err := s.bindAndReleaseNodeBatch(ctx, txStore, storedNodesByFile, batch.nodeBatches, i); err != nil { return err } } diff --git a/internal/app/ingest/workflow/indexer_test.go b/internal/app/ingest/workflow/indexer_test.go index 85c6017..5871a60 100644 --- a/internal/app/ingest/workflow/indexer_test.go +++ b/internal/app/ingest/workflow/indexer_test.go @@ -116,6 +116,7 @@ type recordingGraphStore struct { nextID uint nodesByFP map[string][]graph.Node edges []graph.Edge + upsertedNodeBatches [][]graph.Node upsertedEdges [][]graph.Edge fileSuffixLookupCalls int importFileNodeCalls int @@ -137,6 +138,7 @@ func (r *recordingGraphStore) DeleteGraph(ctx context.Context) error { func (r *recordingGraphStore) UpsertNodes(ctx context.Context, nodes []graph.Node) error { r.record("UpsertNodes") + r.upsertedNodeBatches = append(r.upsertedNodeBatches, append([]graph.Node(nil), nodes...)) for i := range nodes { r.nextID++ nodes[i].ID = r.nextID @@ -207,6 +209,7 @@ func (r *recordingGraphStore) GetNodesByQualifiedNames(ctx context.Context, name } func (r *recordingGraphStore) GetNodesByFiles(ctx context.Context, filePaths []string) (map[string][]graph.Node, error) { + r.record("GetNodesByFiles") set := make(map[string]bool, len(filePaths)) for _, fp := range filePaths { set[fp] = true @@ -2334,7 +2337,7 @@ func Keep() {} t.Fatalf("Build: %v", err) } - want := []string{"DeleteGraph", "UpsertNodes", "GetNodesByFile", "UpsertAnnotation", "UpsertEdges"} + want := []string{"DeleteGraph", "UpsertNodes", "GetNodesByFiles", "UpsertAnnotation", "UpsertEdges"} for i, op := range want[:4] { if len(fakeStore.ops) <= i { t.Fatalf("ops too short: got %v", fakeStore.ops) @@ -2359,6 +2362,81 @@ func Keep() {} } } +func TestFlushBuildBatch_BatchesNodesAndAnnotationLookups(t *testing.T) { + fakeStore := newRecordingGraphStore(t) + released := 0 + svc := &Service{ + onBatchRelease: func([]parsedBuildNodeBatch, int) { + released++ + }, + } + batch := buildPersistBatch{ + files: 2, + nodeBatches: []parsedBuildNodeBatch{ + { + relPath: "first.go", + nodes: []graph.Node{{ + QualifiedName: "sample.First", + FilePath: "first.go", + Kind: graph.NodeKindFunction, + StartLine: 2, + EndLine: 2, + }}, + tsComments: []ingest.CommentBlock{{StartLine: 1, EndLine: 1, Text: "@intent first"}}, + language: "go", + sourceLines: []string{"// @intent first", "func First() {}"}, + }, + { + relPath: "second.go", + nodes: []graph.Node{{ + QualifiedName: "sample.Second", + FilePath: "second.go", + Kind: graph.NodeKindFunction, + StartLine: 2, + EndLine: 2, + }}, + tsComments: []ingest.CommentBlock{{StartLine: 1, EndLine: 1, Text: "@intent second"}}, + language: "go", + sourceLines: []string{"// @intent second", "func Second() {}"}, + }, + }, + } + + if err := svc.flushBuildBatch(context.Background(), fakeStore, &batch); err != nil { + t.Fatalf("flushBuildBatch: %v", err) + } + if got := len(fakeStore.upsertedNodeBatches); got != 1 { + t.Fatalf("node upsert calls = %d, want 1 (ops=%v)", got, fakeStore.ops) + } + if got := len(fakeStore.upsertedNodeBatches[0]); got != 2 { + t.Fatalf("batched node count = %d, want 2", got) + } + countOperation := func(want string) int { + count := 0 + for _, op := range fakeStore.ops { + if op == want { + count++ + } + } + return count + } + if got := countOperation("GetNodesByFiles"); got != 1 { + t.Fatalf("bulk annotation node lookups = %d, want 1 (ops=%v)", got, fakeStore.ops) + } + if got := countOperation("GetNodesByFile"); got != 0 { + t.Fatalf("per-file annotation node lookups = %d, want 0 (ops=%v)", got, fakeStore.ops) + } + if got := countOperation("UpsertAnnotation"); got != 2 { + t.Fatalf("annotation upserts = %d, want 2 (ops=%v)", got, fakeStore.ops) + } + if released != 2 { + t.Fatalf("released batches = %d, want 2", released) + } + if batch.files != 0 || len(batch.nodeBatches) != 0 { + t.Fatalf("batch was not reset: %+v", batch) + } +} + func TestBuild_FlushesLargeBuildInBoundedBatches(t *testing.T) { fakeStore := newRecordingGraphStore(t) svc := &Service{