fix(logger): prevent concurrent console logs from interleaving

logger/console.go

@@ -32,6 +32,12 @@ func (c *consoleWriter) Write(p []byte) (n int, err error) {
 	if err != nil {
 		return n, fmt.Errorf("cannot decode event: %s", err)
 	}
-	e := json.NewEncoder(c.out)
-	return len(p), e.Encode(evt)
+	// buffer each log in-memory to prevent
+	// concurrent logs from interleaving
+	var buf bytes.Buffer
+	if err = json.NewEncoder(&buf).Encode(evt); err != nil {
+		return n, err
+	}
+	_, err = c.out.Write(buf.Bytes())
+	return len(p), err
 }

logger/console_test.go

@@ -3,6 +3,7 @@ package logger
 import (
 	"bytes"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/rs/zerolog"
@@ -31,3 +32,107 @@ func TestConsoleLoggerDuplicateKeys(t *testing.T) {
 		t.Errorf("log event missing key 'level': %s", event)
 	}
 }
+
+type countingWriter struct {
+	bytes.Buffer
+	writes int
+}
+
+func (c *countingWriter) Write(b []byte) (int, error) {
+	c.writes++
+	return c.Buffer.Write(b)
+}
+
+func TestConsoleLoggerWrites(t *testing.T) {
+	cw := &countingWriter{}
+	logger := zerolog.New(&consoleWriter{out: cw}).With().Timestamp().Logger()
+
+	logger.Error().
+		Int("connIndex", 3).
+		Str("error", "context canceled").
+		Str("kind", "http").
+		Msg("Request failed")
+
+	if cw.writes != 1 {
+		t.Fatalf("expected one underlying Write call for the event, got %d", cw.writes)
+	}
+	if !bytes.HasSuffix(cw.Bytes(), []byte("\n")) {
+		t.Fatalf("expected output to end with a newline, got %q", cw.String())
+	}
+	var event map[string]any
+	if err := json.Unmarshal(bytes.TrimSpace(cw.Bytes()), &event); err != nil {
+		t.Fatalf("expected valid JSON output, got %q: %v", cw.String(), err)
+	}
+	if event["message"] != "Request failed" {
+		t.Errorf("unexpected message field: got %v", event["message"])
+	}
+}
+
+// syncedBuffer makes the test safe while preserving per-Write
+// interleaving, like concurrent writes to stderr.
+type syncedBuffer struct {
+	mu     sync.Mutex
+	buffer bytes.Buffer
+}
+
+func (w *syncedBuffer) Write(p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.buffer.Write(p)
+}
+
+func (w *syncedBuffer) Bytes() []byte {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return bytes.Clone(w.buffer.Bytes())
+}
+
+func TestConsoleLoggerConcurrentWrites(t *testing.T) {
+	t.Parallel()
+
+	const (
+		goroutines = 8
+		events     = 5_000
+	)
+
+	buf := &syncedBuffer{}
+	logger := zerolog.New(&consoleWriter{out: buf}).With().Timestamp().Logger()
+
+	var wg sync.WaitGroup
+	for g := range goroutines {
+		wg.Go(func() {
+			for i := 0; i < events; i++ {
+				logger.Error().
+					Int("connIndex", g).
+					Int("event", 0).
+					Str("dest", "https://example.com/api/x").
+					Str("error", "Incoming request ended abruptly: context canceled").
+					Str("type", "http").
+					Msg("Request failed")
+			}
+		})
+	}
+	wg.Wait()
+
+	output := buf.Bytes()
+	lines := bytes.Split(
+		bytes.TrimSuffix(output, []byte("\n")),
+		[]byte("\n"),
+	)
+
+	expected := goroutines * events
+	if len(lines) != expected {
+		t.Fatalf("expected %d JSON lines, got %d", expected, len(lines))
+	}
+
+	invalid := 0
+	for _, line := range lines {
+		var event map[string]any
+		if err := json.Unmarshal(line, &event); err != nil {
+			invalid++
+		}
+	}
+	if invalid > 0 {
+		t.Errorf("%d of %d output lines are invalid JSON", invalid, len(lines))
+	}
+}