Function processComponentsInParallel

cmd/validate/vsa.go:490–562

processComponentsInParallel processes all components using parallel workers

(ctx context.Context, components []app.SnapshotComponent, data *validateVSAData)

// processComponentsInParallel processes all components using parallel workers
func processComponentsInParallel(ctx context.Context, components []app.SnapshotComponent, data *validateVSAData) ([]vsa.ComponentResult, error) {
	numComponents := len(components)
	numWorkers := data.workers

	printVSAInfo(os.Stdout, fmt.Sprintf("Found %d components in snapshot", numComponents))
	printVSAInfo(os.Stdout, fmt.Sprintf("=== Processing Components in Parallel (%d workers) ===", numWorkers))

	// Add timeout for parallel processing to prevent hanging
	ctx, cancel := context.WithTimeout(ctx, DefaultTimeoutDuration)
	defer cancel()

	// Set up parallel processing infrastructure
	jobs := make(chan app.SnapshotComponent, numComponents)
	results := make(chan vsa.ComponentResult, numComponents)

	// Use WaitGroup to track worker completion
	var wg sync.WaitGroup

	// Start worker goroutines
	for i := 0; i < numWorkers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			worker(jobs, results, ctx, data)
		}()
	}

	// Close results channel after all workers finish
	go func() {
		wg.Wait()
		close(results)
	}()

	// Send all components to workers
	for _, component := range components {
		select {
		case jobs <- component:
			// Component sent successfully
		case <-ctx.Done():
			close(jobs)
			// Don't wait for workers - context cancellation will signal them to stop
			// and the results channel will be closed by the goroutine after workers finish
			return nil, fmt.Errorf("parallel processing timeout after %v: %w", DefaultTimeoutDuration, ctx.Err())
		}
	}
	close(jobs)

	// Collect results with timeout handling
	var allResults []vsa.ComponentResult
	for i := 0; i < numComponents; i++ {
		select {
		case result, ok := <-results:
			if !ok {
				// Channel closed, but we haven't received all results
				// This shouldn't happen normally, but handle gracefully
				return allResults, fmt.Errorf("results channel closed prematurely (received %d/%d results)", len(allResults), numComponents)
			}
			allResults = append(allResults, result)