fix(output): synchronize concurrent writes to the prefixed writer

internal/output/output_test.go

@@ -5,6 +5,8 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"strings"
+	"sync"
 	"testing"
 
 	"github.com/fatih/color"
@@ -190,3 +192,36 @@ func TestPrefixedWithColor(t *testing.T) {
 		}
 	})
 }
+
+func TestPrefixedConcurrentWrites(t *testing.T) {
+	t.Parallel()
+
+	var b bytes.Buffer
+	l := &logger.Logger{Color: false}
+	var o output.Output = output.NewPrefixed(l)
+	stdOut, stdErr, cleanup := o.WrapWriter(&b, &b, "prefix", nil)
+
+	const lines = 1000
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		for i := range lines {
+			fmt.Fprintf(stdOut, "stdout-%d\n", i)
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		for i := range lines {
+			fmt.Fprintf(stdErr, "stderr-%d\n", i)
+		}
+	}()
+	wg.Wait()
+	require.NoError(t, cleanup(nil))
+
+	out := strings.Split(strings.TrimSuffix(b.String(), "\n"), "\n")
+	require.Len(t, out, lines*2)
+	for _, line := range out {
+		assert.True(t, strings.HasPrefix(line, "[prefix] "))
+	}
+}

internal/output/prefixed.go

@@ -38,9 +38,13 @@ type prefixWriter struct {
 	prefixed *Prefixed
 	prefix   string
 	buff     bytes.Buffer
+	mutex    sync.Mutex
 }
 
 func (pw *prefixWriter) Write(p []byte) (int, error) {
+	pw.mutex.Lock()
+	defer pw.mutex.Unlock()
+
 	n, err := pw.buff.Write(p)
 	if err != nil {
 		return n, err
@@ -50,6 +54,9 @@ func (pw *prefixWriter) Write(p []byte) (int, error) {
 }
 
 func (pw *prefixWriter) close() error {
+	pw.mutex.Lock()
+	defer pw.mutex.Unlock()
+
 	return pw.writeOutputLines(true)
 }