From 1d4d8bc8127beb4b8ac58e122f6b5bd7664fc00c Mon Sep 17 00:00:00 2001
From: AdamMagued <adamismailmageud@gmail.com>
Date: Fri, 9 Oct 2026 09:27:58 +0300
Subject: [PATCH] fix(stream): allow concurrent execution when max goroutines
 is unbounded

Signed-off-by: AdamMagued <adamismailmageud@gmail.com>
---
 stream/stream.go      | 71 ++++++++++++++++++++++++++++++++++++++++---
 stream/stream_test.go | 63 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 129 insertions(+), 5 deletions(-)

diff --git a/stream/stream.go b/stream/stream.go
index 6b11e90..996102c 100644
--- a/stream/stream.go
+++ b/stream/stream.go
@@ -38,7 +38,7 @@ func New() *Stream {
 type Stream struct {
 	pool             pool.Pool
 	callbackerHandle conc.WaitGroup
-	queue            chan callbackCh
+	queue            *callbackQueue
 
 	initOnce sync.Once
 }
@@ -66,7 +66,7 @@ func (s *Stream) Go(f Task) {
 	ch := getCh()
 
 	// Queue the channel for the callbacker.
-	s.queue <- ch
+	s.queue.push(ch)
 
 	// Submit the task for execution.
 	s.pool.Go(func() {
@@ -94,7 +94,7 @@ func (s *Stream) Wait() {
 	// Defer the callbacker cleanup so that it occurs even in the case
 	// that one of the tasks panics and is propagated up by s.pool.Wait().
 	defer func() {
-		close(s.queue)
+		s.queue.close()
 		s.callbackerHandle.Wait()
 	}()
 
@@ -109,7 +109,7 @@ func (s *Stream) WithMaxGoroutines(n int) *Stream {
 
 func (s *Stream) init() {
 	s.initOnce.Do(func() {
-		s.queue = make(chan callbackCh, s.pool.MaxGoroutines()+1)
+		s.queue = newCallbackQueue()
 
 		// Start the callbacker.
 		s.callbackerHandle.Go(s.callbacker)
@@ -123,7 +123,12 @@ func (s *Stream) callbacker() {
 	defer panicCatcher.Repanic()
 
 	// For every scheduled task, read that tasks channel from the queue.
-	for callbackCh := range s.queue {
+	for {
+		callbackCh, ok := s.queue.pop()
+		if !ok {
+			break
+		}
+
 		// Wait for the task to complete and get its callback from the channel.
 		callback := <-callbackCh
 
@@ -137,6 +142,62 @@ func (s *Stream) callbacker() {
 	}
 }
 
+type callbackQueue struct {
+	mu     sync.Mutex
+	cond   *sync.Cond
+	items  []callbackCh
+	head   int
+	closed bool
+}
+
+func newCallbackQueue() *callbackQueue {
+	q := &callbackQueue{}
+	q.cond = sync.NewCond(&q.mu)
+	return q
+}
+
+func (q *callbackQueue) push(ch callbackCh) {
+	q.mu.Lock()
+	q.items = append(q.items, ch)
+	q.cond.Signal()
+	q.mu.Unlock()
+}
+
+func (q *callbackQueue) pop() (callbackCh, bool) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	for q.head == len(q.items) && !q.closed {
+		q.cond.Wait()
+	}
+
+	if q.head == len(q.items) && q.closed {
+		return nil, false
+	}
+
+	ch := q.items[q.head]
+	q.items[q.head] = nil
+	q.head++
+
+	if q.head == len(q.items) {
+		q.items = q.items[:0]
+		q.head = 0
+	} else if q.head > 1024 && q.head*2 >= len(q.items) {
+		copy(q.items, q.items[q.head:])
+		q.items = q.items[:len(q.items)-q.head]
+		q.head = 0
+	}
+
+	return ch, true
+}
+
+func (q *callbackQueue) close() {
+	q.mu.Lock()
+	q.closed = true
+	q.cond.Broadcast()
+	q.mu.Unlock()
+}
+
 type callbackCh chan func()
 
 var callbackChPool = sync.Pool{
diff --git a/stream/stream_test.go b/stream/stream_test.go
index 9f5bce1..11ea5da 100644
--- a/stream/stream_test.go
+++ b/stream/stream_test.go
@@ -133,6 +133,69 @@ func TestStream(t *testing.T) {
 		}
 		require.Panics(t, s.Wait)
 	})
+
+	t.Run("unbounded concurrent execution", func(t *testing.T) {
+		t.Parallel()
+		s := stream.New()
+		const taskCount = 10
+		started := make(chan struct{}, taskCount)
+		release := make(chan struct{})
+
+		for i := 0; i < taskCount; i++ {
+			s.Go(func() stream.Callback {
+				started <- struct{}{}
+				<-release
+				return nil
+			})
+		}
+
+		for i := 0; i < taskCount; i++ {
+			select {
+			case <-started:
+			case <-time.After(5 * time.Second):
+				t.Fatalf("timed out waiting for task %d to start concurrently", i)
+			}
+		}
+
+		close(release)
+		s.Wait()
+	})
+
+	t.Run("unbounded ordered callbacks with inverted completion", func(t *testing.T) {
+		t.Parallel()
+		s := stream.New()
+		const taskCount = 10
+		var results []int
+
+		for i := 0; i < taskCount; i++ {
+			i := i
+			s.Go(func() stream.Callback {
+				// Earlier tasks take longer so that completion order is inverted.
+				sleepDuration := time.Duration(taskCount-i) * 5 * time.Millisecond
+				time.Sleep(sleepDuration)
+				return func() {
+					results = append(results, i)
+				}
+			})
+		}
+
+		s.Wait()
+
+		expected := make([]int, taskCount)
+		for i := 0; i < taskCount; i++ {
+			expected[i] = i
+		}
+		require.Equal(t, expected, results)
+	})
+
+	t.Run("unbounded panic propagation", func(t *testing.T) {
+		t.Parallel()
+		s := stream.New()
+		s.Go(func() stream.Callback {
+			panic("panic in unbounded task")
+		})
+		require.Panics(t, s.Wait)
+	})
 }
 
 func BenchmarkStream(b *testing.B) {