fix: Rewrite regristy concurrency to avoid deadlocks

Author: matzehuels

internal/cli/parse.go

@@ -27,7 +27,7 @@ type parseOpts struct {
 type parserFactory func() (source.Parser, error)
 
 func newParseCmd() *cobra.Command {
-	opts := parseOpts{maxDepth: 10, maxNodes: 100}
+	opts := parseOpts{maxDepth: 10, maxNodes: 5000}
 
 	cmd := &cobra.Command{
 		Use:   "parse",

pkg/source/registry.go

@@ -30,7 +30,7 @@ type RepoInfo struct {
 	Version      string
 	ProjectURLs  map[string]string
 	HomePage     string
-	ManifestFile string // e.g., "Cargo.toml", "pyproject.toml", "package.json"
+	ManifestFile string
 }
 
 type Options struct {
@@ -68,18 +68,6 @@ type PackageInfo interface {
 
 type fetchFunc[T PackageInfo] func(ctx context.Context, name string, refresh bool) (T, error)
 
-type job struct {
-	name  string
-	depth int
-}
-
-type result[T PackageInfo] struct {
-	name  string
-	info  T
-	depth int
-	err   error
-}
-
 func Parse[T PackageInfo](ctx context.Context, root string, opts Options, fetch fetchFunc[T]) (*dag.DAG, error) {
 	opts = opts.withDefaults()
 
@@ -90,13 +78,26 @@ func Parse[T PackageInfo](ctx context.Context, root string, opts Options, fetch
 		g:       dag.New(nil),
 		visited: make(map[string]bool),
 		meta:    make(map[string]map[string]any),
-		jobs:    make(chan job, numWorkers),
-		results: make(chan result[T], numWorkers),
+		jobs:    make(chan job, numWorkers*2),
+		results: make(chan result[T], numWorkers*2),
+		done:    make(chan struct{}),
 	}
 
 	return p.parse(root)
 }
 
+type job struct {
+	name  string
+	depth int
+}
+
+type result[T PackageInfo] struct {
+	name  string
+	info  T
+	depth int
+	err   error
+}
+
 type parser[T PackageInfo] struct {
 	ctx   context.Context
 	opts  Options
@@ -108,30 +109,40 @@ type parser[T PackageInfo] struct {
 
 	jobs    chan job
 	results chan result[T]
+	done    chan struct{}
+
+	mu        sync.Mutex
+	inflight  int64
+	nodeCount int32
+}
 
-	mu     sync.Mutex
-	active int
+func (p *parser[T]) adjustInflight(delta int64) {
+	p.mu.Lock()
+	p.inflight += delta
+	isDone := p.inflight == 0
+	p.mu.Unlock()
+
+	if isDone {
+		close(p.done)
+	}
 }
 
 func (p *parser[T]) parse(root string) (*dag.DAG, error) {
-	var wg sync.WaitGroup
+	var workerWg sync.WaitGroup
 	for range numWorkers {
-		wg.Add(1)
+		workerWg.Add(1)
 		go func() {
-			defer wg.Done()
+			defer workerWg.Done()
 			p.worker()
 		}()
 	}
 
-	go func() {
-		wg.Wait()
-		close(p.results)
-	}()
-
 	p.submit(job{name: root, depth: 0})
 
 	rootErr := p.processResults(root)
+
 	close(p.jobs)
+	workerWg.Wait()
 
 	if rootErr != nil {
 		return nil, rootErr
@@ -144,92 +155,128 @@ func (p *parser[T]) parse(root string) (*dag.DAG, error) {
 func (p *parser[T]) worker() {
 	for j := range p.jobs {
 		if p.ctx.Err() != nil {
-			return
+			p.adjustInflight(-1) // job cancelled
+			continue
 		}
 		info, err := p.fetch(p.ctx, j.name, p.opts.Refresh)
 		p.results <- result[T]{name: j.name, info: info, depth: j.depth, err: err}
 	}
 }
 
-func (p *parser[T]) submit(j job) {
+func (p *parser[T]) submit(j job) bool {
 	p.mu.Lock()
 	if p.visited[j.name] {
 		p.mu.Unlock()
-		return
+		return false
 	}
 	p.visited[j.name] = true
-	p.active++
+	p.inflight++
 	p.mu.Unlock()
 
 	p.jobs <- j
+	return true
 }
 
 func (p *parser[T]) processResults(root string) error {
-	for r := range p.results {
-		p.decrementActive()
-
-		if r.err != nil {
-			if r.name == root {
-				return r.err
+	for {
+		select {
+		case r := <-p.results:
+			if err := p.handleResult(r, root); err != nil {
+				return err
 			}
-			p.opts.Logger("failed to fetch %s: %v", r.name, r.err)
-		} else {
-			p.processResult(r)
+
+		case <-p.done:
+			return nil
+
+		case <-p.ctx.Done():
+			return p.ctx.Err()
 		}
+	}
+}
 
-		if p.isDone() {
-			break
+func (p *parser[T]) handleResult(r result[T], root string) error {
+	defer p.adjustInflight(-1)
+
+	if r.err != nil {
+		if r.name == root {
+			return r.err
 		}
+		p.opts.Logger("failed to fetch %s: %v", r.name, r.err)
+		return nil
 	}
+
+	p.addNode(r)
+	p.submitDependencies(r)
 	return nil
 }
 
-func (p *parser[T]) processResult(r result[T]) {
+func (p *parser[T]) addNode(r result[T]) {
 	_ = p.g.AddNode(dag.Node{ID: r.name})
 
+	p.mu.Lock()
+	p.nodeCount++
+	p.mu.Unlock()
+
 	meta := enrichMetadata(p.ctx, r.info, p.opts)
-	p.storeMeta(r.name, meta)
+	if len(meta) > 0 {
+		p.mu.Lock()
+		p.meta[r.name] = meta
+		p.mu.Unlock()
+	}
+}
 
-	if r.depth >= p.opts.MaxDepth || len(p.visited) >= p.opts.MaxNodes {
+func (p *parser[T]) submitDependencies(r result[T]) {
+	if r.depth >= p.opts.MaxDepth {
 		return
 	}
 
-	for _, dep := range r.info.GetDependencies() {
-		_ = p.g.AddNode(dag.Node{ID: dep})
-		_ = p.g.AddEdge(dag.Edge{From: r.name, To: dep})
-		p.submit(job{name: dep, depth: r.depth + 1})
+	deps := r.info.GetDependencies()
+	if len(deps) == 0 {
+		return
 	}
-}
 
-func (p *parser[T]) applyMetadata() {
+	// Add edges and collect jobs
 	p.mu.Lock()
-	defer p.mu.Unlock()
-	for id, m := range p.meta {
-		if n, ok := p.g.Node(id); ok {
-			n.Meta = m
+	nodeCount := p.nodeCount
+	p.mu.Unlock()
+
+	var toSubmit []job
+	for _, dep := range deps {
+		_ = p.g.AddNode(dag.Node{ID: dep})
+		_ = p.g.AddEdge(dag.Edge{From: r.name, To: dep})
+
+		if int(nodeCount) < p.opts.MaxNodes {
+			toSubmit = append(toSubmit, job{name: dep, depth: r.depth + 1})
 		}
 	}
-}
 
-func (p *parser[T]) decrementActive() {
+	if len(toSubmit) == 0 {
+		return
+	}
+
+	// Reserve a slot for the async submitter BEFORE spawning it.
+	// This prevents processResults from exiting prematurely.
 	p.mu.Lock()
-	p.active--
+	p.inflight++
 	p.mu.Unlock()
+
+	go func() {
+		defer p.adjustInflight(-1) // release slot when done submitting
+
+		for _, j := range toSubmit {
+			p.submit(j)
+		}
+	}()
 }
 
-func (p *parser[T]) isDone() bool {
+func (p *parser[T]) applyMetadata() {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	return p.active == 0
-}
-
-func (p *parser[T]) storeMeta(name string, meta map[string]any) {
-	if len(meta) == 0 {
-		return
+	for id, m := range p.meta {
+		if n, ok := p.g.Node(id); ok {
+			n.Meta = m
+		}
 	}
-	p.mu.Lock()
-	p.meta[name] = meta
-	p.mu.Unlock()
 }
 
 func enrichMetadata(ctx context.Context, info PackageInfo, opts Options) map[string]any {