fix(envd): fix data race in fan-out when subscriber is removed mid-iteration

run() copied only the slice header under RLock, sharing the backing
array with remove(). When remove() shifted elements in-place via
append(channels[:i], channels[i+1:]...), the fan-out's stale snapshot
would skip one subscriber and deliver to another twice — silent
stdout/stderr corruption. Deep-copy the slice under RLock so the
iteration is immune to concurrent mutations.

Fixes: a67f983 ("fix(envd): fix fan-out deadlock when process subscriber disconnects (#2579)")

Author: arkamar

packages/envd/internal/services/process/handler/multiplex.go

@@ -54,7 +54,7 @@ func NewMultiplexedChannel[T any](buffer int) *MultiplexedChannel[T] {
 func (m *MultiplexedChannel[T]) run() {
 	for v := range m.Source {
 		m.mu.RLock()
-		subs := m.channels
+		subs := append([]*subscriber[T](nil), m.channels...)
 		m.mu.RUnlock()
 
 		for _, s := range subs {

packages/envd/internal/services/process/handler/multiplex_test.go

@@ -312,3 +312,90 @@ func TestMultiplexedChannel_NoGoroutineLeakOnAbandon(t *testing.T) { //nolint:pa
 			"before=%d after=%d (expected ~0)", leaked, wedges, before, after,
 	)
 }
+
+// Regression: removing a subscriber while the fan-out is mid-iteration
+// must not corrupt delivery to the remaining subscribers.  Before the
+// fix, run() copied only the slice header (sharing the backing array),
+// so remove()'s in-place shift could cause duplicates and skips.
+// Run with -race to verify no data race.
+func TestMultiplexedChannel_RemoveDuringFanOutDoesNotCorrupt(t *testing.T) {
+	t.Parallel()
+
+	const iterations = 500
+	const values = 20
+
+	for iter := range iterations {
+		m := NewMultiplexedChannel[int](values)
+
+		chA, cancelA := m.Fork()
+		chB, cancelB := m.Fork()
+		chC, cancelC := m.Fork()
+
+		// Pump values into the buffered Source — all fit without blocking.
+		for i := 1; i <= values; i++ {
+			m.Source <- i
+		}
+
+		// A reads one value then cancels — triggers remove() while
+		// the fan-out is delivering to the remaining subscribers.
+		// No drain needed: after cancel the fan-out skips A via <-s.done.
+		go func() {
+			<-chA
+			cancelA()
+		}()
+
+		// B and C drain everything.
+		bDone := make(chan []int, 1)
+		cDone := make(chan []int, 1)
+		go func() {
+			var got []int
+			for v := range chB {
+				got = append(got, v)
+			}
+			bDone <- got
+		}()
+		go func() {
+			var got []int
+			for v := range chC {
+				got = append(got, v)
+			}
+			cDone <- got
+		}()
+
+		// Shut down. close(Source) causes run() to exit and close
+		// all remaining subscriber channels (B and C), which lets
+		// the drainer goroutines finish and send on bDone/cDone.
+		close(m.Source)
+
+		// Collect results. B and C channels are closed by run()'s
+		// cleanup, so the drainers will terminate. Cancel only
+		// after collecting to avoid racing with run()'s cleanup.
+		bGot := <-bDone
+		cGot := <-cDone
+
+		cancelB()
+		cancelC()
+
+		// B and C must each receive all values exactly once.
+		if len(bGot) != values {
+			t.Errorf("iter %d: B got %d values, want %d: %v", iter, len(bGot), values, bGot)
+		}
+		if len(cGot) != values {
+			t.Errorf("iter %d: C got %d values, want %d: %v", iter, len(cGot), values, cGot)
+		}
+
+		cCount := map[int]int{}
+		for _, v := range cGot {
+			cCount[v]++
+		}
+		for v, n := range cCount {
+			if n > 1 {
+				t.Errorf("iter %d: C got value %d %d times (duplicate delivery)", iter, v, n)
+			}
+		}
+
+		if t.Failed() {
+			break
+		}
+	}
+}