fix: process pool concurrency (#1358)

* remove telemetry from bindings

* optimise process pool

* optimise test cases

* fix test cases

* revert telemetry related changes

Author: krishankumar01

foundation/application.go

@@ -128,6 +128,8 @@ func (r *Application) Context() context.Context {
 	return r.ctx
 }
 
+// GetJson get the JSON implementation.
+// DEPRECATED, use Json instead.
 func (r *Application) GetJson() foundation.Json {
 	return r.json
 }

process/pool.go

@@ -79,28 +79,26 @@ func (r *PoolBuilder) WithSpinner(message ...string) contractsprocess.PoolBuilde
 	return r
 }
 
-type job struct {
-	id      int
-	command *PoolCommand
-}
-
 type result struct {
 	key string
 	res contractsprocess.Result
 }
 
-// start initiates the execution of all configured commands concurrently but does not wait for them to complete.
-// It orchestrates a pool of worker goroutines to process commands up to the specified concurrency limit.
+// start initiates the execution of all configured commands concurrently based on the
+// concurrency limit, but does not wait for them to complete.
 //
-// This method is non-blocking. It returns a RunningPool instance immediately, which can be used to
-// wait for the completion of all processes and retrieve their results.
+// This method is non-blocking. It returns a RunningPool instance immediately, which
+// can be used to wait for the completion of all processes and retrieve their results.
 //
 // The core concurrency pattern is as follows:
 //  1. A job channel (`jobCh`) distributes commands to a pool of worker goroutines.
-//  2. A result channel (`resultCh`) collects the outcome of each command from a dedicated waiter goroutine.
-//  3. A separate "collector" goroutine safely populates the final results map from the result channel.
-//  4. WaitGroups synchronize the completion of all workers and the collection of all results
-//     before the entire operation is marked as "done".
+//  2. Workers pick up a job, start the process, and wait synchronously for it to finish.
+//     This synchronous wait ensures the concurrency limit is strictly respected.
+//  3. A result channel (`resultCh`) collects the outcome (success/failure) of each command.
+//  4. A separate "collector" goroutine safely populates the RunningPool's internal map
+//     from the result channel to avoid concurrent map write panics.
+//  5. A background orchestrator waits for all workers and results to finish, then
+//     cleanly closes resources and signals the `done` channel.
 func (r *PoolBuilder) start(configurer func(contractsprocess.Pool)) (contractsprocess.RunningPool, error) {
 	if configurer == nil {
 		return nil, errors.ProcessPoolNilConfigurer
@@ -118,25 +116,28 @@ func (r *PoolBuilder) start(configurer func(contractsprocess.Pool)) (contractspr
 	var cancel context.CancelFunc
 	if r.timeout > 0 {
 		ctx, cancel = context.WithTimeout(ctx, r.timeout)
+	} else {
+		ctx, cancel = context.WithCancel(ctx)
 	}
 
-	concurrency := r.concurrency
-	if concurrency <= 0 || concurrency > len(commands) {
-		concurrency = len(commands)
-	}
-
-	jobCh := make(chan job, len(commands))
+	jobCh := make(chan *PoolCommand, len(commands))
 	resultCh := make(chan result, len(commands))
 	done := make(chan struct{})
 
-	results := make(map[string]contractsprocess.Result, len(commands))
-	runningProcesses := make([]contractsprocess.Running, len(commands))
 	keys := make([]string, len(commands))
+	for i, cmd := range commands {
+		keys[i] = cmd.key
+	}
+
+	runningPool := NewRunningPool(ctx, cancel, keys, done, r.loading, r.loadingMessage)
+
+	concurrency := r.concurrency
+	if concurrency <= 0 || concurrency > len(commands) {
+		concurrency = len(commands)
+	}
 
 	var resultsWg sync.WaitGroup
 	var workersWg sync.WaitGroup
-	var startsWg sync.WaitGroup
-	var mu sync.Mutex
 
 	// The results collector goroutine centralizes writing to the results map
 	// to avoid race conditions, as map writes are not concurrent-safe.
@@ -145,9 +146,7 @@ func (r *PoolBuilder) start(configurer func(contractsprocess.Pool)) (contractspr
 	go func() {
 		for i := 0; i < len(commands); i++ {
 			rc := <-resultCh
-			mu.Lock()
-			results[rc.key] = rc.res
-			mu.Unlock()
+			runningPool.setResult(rc.key, rc.res)
 			resultsWg.Done()
 		}
 	}()
@@ -156,8 +155,16 @@ func (r *PoolBuilder) start(configurer func(contractsprocess.Pool)) (contractspr
 		workersWg.Add(1)
 		go func() {
 			defer workersWg.Done()
-			for currentJob := range jobCh {
-				command := currentJob.command
+			for command := range jobCh {
+				if ctx.Err() != nil {
+					// If the pool was stopped (Stop() called or timeout reached), we skip execution.
+					// We must still send a result to ensure resultsWg decrements correctly.
+					resultCh <- result{
+						key: command.key,
+						res: NewResult(ctx.Err(), -1, "", "", ""),
+					}
+					continue
+				}
 				cmdCtx := command.ctx
 				if cmdCtx == nil {
 					cmdCtx = ctx
@@ -184,36 +191,19 @@ func (r *PoolBuilder) start(configurer func(contractsprocess.Pool)) (contractspr
 				if err != nil {
 					resultCh <- result{key: command.key, res: NewResult(err, -1, command.name, "", "")}
 				} else {
-					mu.Lock()
-					runningProcesses[currentJob.id] = run
-					mu.Unlock()
-
-					// Launch a dedicated goroutine to wait for the process to finish.
-					// This prevents the worker from being blocked by a long-running process
-					// and allows it to immediately pick up the next job from jobCh.
-					go func(p contractsprocess.Running, k string) {
-						res := p.Wait()
-						resultCh <- result{key: k, res: res}
-					}(run, command.key)
+					runningPool.setProcess(command.key, run)
+					res := run.Wait()
+					resultCh <- result{key: command.key, res: res}
 				}
-
-				// Signal that this process has completed its start attempt
-				startsWg.Done()
 			}
 		}()
 	}
 
-	startsWg.Add(len(commands))
-	for i, command := range commands {
-		keys[i] = command.key
-		jobCh <- job{id: i, command: command}
+	for _, command := range commands {
+		jobCh <- command
 	}
 	close(jobCh)
 
-	// Wait for all processes to complete their start attempts before returning.
-	// This ensures the runningProcesses slice is fully populated and safe to access.
-	startsWg.Wait()
-
 	// This goroutine orchestrates the clean shutdown. It waits for all workers
 	// to finish processing jobs, then waits for all results to be collected.
 	// Finally, it cancels the context (if a timeout was set) and signals
@@ -227,7 +217,7 @@ func (r *PoolBuilder) start(configurer func(contractsprocess.Pool)) (contractspr
 		close(done)
 	}()
 
-	return NewRunningPool(ctx, runningProcesses, keys, cancel, results, done, r.loading, r.loadingMessage), nil
+	return runningPool, nil
 }
 
 type Pool struct {

process/pool_unix_test.go

@@ -259,12 +259,12 @@ func TestPool_Concurrency_Unix(t *testing.T) {
 
 func TestPool_OnOutput_Unix(t *testing.T) {
 	t.Run("captures output via callback", func(t *testing.T) {
-		mx := sync.Mutex{}
+		var mu sync.Mutex
 		outputs := make(map[string][]string)
 		builder := NewPool().OnOutput(func(typ contractsprocess.OutputType, line []byte, key string) {
-			mx.Lock()
+			mu.Lock()
 			outputs[key] = append(outputs[key], string(line))
-			mx.Unlock()
+			mu.Unlock()
 		})
 
 		rp, err := builder.Pool(func(p contractsprocess.Pool) {
@@ -285,10 +285,13 @@ func TestPool_OnOutput_Unix(t *testing.T) {
 	})
 
 	t.Run("distinguishes stdout and stderr", func(t *testing.T) {
+		var mu sync.Mutex
 		stdoutLines := make(map[string][]string)
 		stderrLines := make(map[string][]string)
 
 		builder := NewPool().OnOutput(func(typ contractsprocess.OutputType, line []byte, key string) {
+			mu.Lock()
+			defer mu.Unlock()
 			if typ == contractsprocess.OutputTypeStdout {
 				stdoutLines[key] = append(stdoutLines[key], string(line))
 			} else {

process/pool_windows_test.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"os"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -257,11 +258,13 @@ func TestPool_Concurrency_Windows(t *testing.T) {
 }
 
 func TestPool_OnOutput_Windows(t *testing.T) {
-	// TODO: fix sync issue when writing concurrently
 	t.Run("captures output via callback", func(t *testing.T) {
+		var mu sync.Mutex
 		outputs := make(map[string][]string)
 		builder := NewPool().OnOutput(func(typ contractsprocess.OutputType, line []byte, key string) {
+			mu.Lock()
 			outputs[key] = append(outputs[key], string(line))
+			mu.Unlock()
 		})
 
 		rp, err := builder.Pool(func(p contractsprocess.Pool) {
@@ -282,10 +285,13 @@ func TestPool_OnOutput_Windows(t *testing.T) {
 	})
 
 	t.Run("distinguishes stdout and stderr", func(t *testing.T) {
+		var mu sync.Mutex
 		stdoutLines := make(map[string][]string)
 		stderrLines := make(map[string][]string)
 
 		builder := NewPool().OnOutput(func(typ contractsprocess.OutputType, line []byte, key string) {
+			mu.Lock()
+			defer mu.Unlock()
 			if typ == contractsprocess.OutputTypeStdout {
 				stdoutLines[key] = append(stdoutLines[key], string(line))
 			} else {
@@ -414,19 +420,17 @@ func TestPool_SignalHandling_Windows(t *testing.T) {
 	t.Run("forwards signals to child processes", func(t *testing.T) {
 		builder := NewPool()
 		rp, err := builder.Pool(func(p contractsprocess.Pool) {
-			// Set up a PowerShell script that can handle CTRL_BREAK_EVENT (mapped from os.Interrupt)
-			p.Command("powershell", "-Command", "$global:interrupted = $false; [console]::TreatControlCAsInput = $true; $handler = [Console]::CancelKeyPress; [Console]::CancelKeyPress = { $global:interrupted = $true; Write-Output 'caught'; $_.Cancel = $true }; Start-Sleep -Seconds 5; if ($global:interrupted) { exit 0 } else { exit 1 }").As("trap")
+			// Start a long sleep that would normally take 10 seconds
+			p.Command("powershell", "-Command", "Start-Sleep -Seconds 10").As("long-running")
 		}).Start()
 		assert.NoError(t, err)
 
-		time.Sleep(100 * time.Millisecond)
-		// Send os.Interrupt which should be mapped to CTRL_BREAK_EVENT on Windows
-		err = rp.Signal(os.Interrupt)
-		// We don't assert on the error because Windows process signal behavior can be inconsistent
+		time.Sleep(1 * time.Second)
+		err = rp.Signal(os.Kill)
+		assert.NoError(t, err)
 
 		results := rp.Wait()
-		// The important thing is that the process completes
-		assert.Contains(t, results, "trap")
+		assert.True(t, results["long-running"].Failed())
 	})
 }
 

process/running_pool.go

@@ -3,6 +3,7 @@ package process
 import (
 	"context"
 	"os"
+	"sync"
 	"time"
 
 	contractsprocess "github.com/goravel/framework/contracts/process"
@@ -11,46 +12,50 @@ import (
 var _ contractsprocess.RunningPool = (*RunningPool)(nil)
 
 type RunningPool struct {
+	mu             sync.RWMutex
 	ctx            context.Context
-	running        []contractsprocess.Running
-	keys           []string
 	cancel         context.CancelFunc
+	done           chan struct{}
+	processes      map[string]contractsprocess.Running
+	results        map[string]contractsprocess.Result
+	keys           []string
 	loading        bool
 	loadingMessage string
-	results        map[string]contractsprocess.Result
-	done           chan struct{}
 }
 
 func NewRunningPool(
 	ctx context.Context,
-	running []contractsprocess.Running,
-	keys []string,
 	cancel context.CancelFunc,
-	results map[string]contractsprocess.Result,
+	keys []string,
 	done chan struct{},
 	loading bool,
 	loadingMessage string,
 ) *RunningPool {
+	processes := make(map[string]contractsprocess.Running, len(keys))
+	results := make(map[string]contractsprocess.Result, len(keys))
+
 	return &RunningPool{
 		ctx:            ctx,
-		running:        running,
-		keys:           keys,
 		cancel:         cancel,
+		keys:           keys,
+		done:           done,
+		processes:      processes,
+		results:        results,
 		loading:        loading,
 		loadingMessage: loadingMessage,
-		results:        results,
-		done:           done,
 	}
 }
 
 func (r *RunningPool) PIDs() map[string]int {
-	m := make(map[string]int, len(r.running))
-	for i, proc := range r.running {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	m := make(map[string]int, len(r.keys))
+	for _, key := range r.keys {
 		pid := 0
-		if proc != nil {
+		if proc, ok := r.processes[key]; ok {
 			pid = proc.PID()
 		}
-		m[r.keys[i]] = pid
+		m[key] = pid
 	}
 	return m
 }
@@ -69,18 +74,27 @@ func (r *RunningPool) Done() <-chan struct{} {
 }
 
 func (r *RunningPool) Wait() map[string]contractsprocess.Result {
-	if err := r.spinner(func() error {
+	_ = r.spinner(func() error {
 		<-r.Done()
 		return nil
-	}); err != nil {
-		return r.results
-	}
+	})
+
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
 	return r.results
 }
 
 func (r *RunningPool) Stop(timeout time.Duration, sig ...os.Signal) error {
+	if r.cancel != nil {
+		r.cancel()
+	}
+
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
 	var firstErr error
-	for _, proc := range r.running {
+	for _, proc := range r.processes {
 		if proc == nil {
 			continue
 		}
@@ -92,8 +106,10 @@ func (r *RunningPool) Stop(timeout time.Duration, sig ...os.Signal) error {
 }
 
 func (r *RunningPool) Signal(sig os.Signal) error {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
 	var firstErr error
-	for _, proc := range r.running {
+	for _, proc := range r.processes {
 		if proc == nil {
 			continue
 		}
@@ -104,6 +120,18 @@ func (r *RunningPool) Signal(sig os.Signal) error {
 	return firstErr
 }
 
+func (r *RunningPool) setProcess(key string, proc contractsprocess.Running) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.processes[key] = proc
+}
+
+func (r *RunningPool) setResult(key string, res contractsprocess.Result) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.results[key] = res
+}
+
 func (r *RunningPool) spinner(fn func() error) error {
 	loadingMessage := r.loadingMessage
 	if loadingMessage == "" {