fix(logger): hand psrpc the service logger so its bus failures are logged

.changeset/psrpc-service-logger.md

@@ -0,0 +1,6 @@
+---
+"github.com/livekit/protocol": patch
+"@livekit/protocol": patch
+---
+
+Hand psrpc the service logger in SetLogger so its bus errors are logged

logger/logger.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/go-logr/logr"
 	"github.com/go-logr/logr/funcr"
+	"github.com/livekit/psrpc"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 
@@ -59,6 +60,9 @@ func SetLogger(l Logger, name string) {
 	defaultLogger = l.WithName(name)
 	// pkg wrapper needs to drop two levels of depth
 	pkgLogger = l.WithCallDepth(1).WithName(name)
+	// psrpc reports its own bus failures (a dropped Redis subscription, a failed publish) only
+	// through its own package logger, which discards until one is set.
+	psrpc.SetLogger(logr.FromSlogHandler(ToSlogHandler(defaultLogger.WithComponent("psrpc"))))
 }
 
 func Debugw(msg string, keysAndValues ...any) {

logger/logger_test.go

@@ -1,13 +1,21 @@
 package logger
 
 import (
+	"bytes"
+	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"net"
 	"os"
 	"runtime"
 	"strings"
+	"sync/atomic"
 	"testing"
+	"time"
 
+	"github.com/livekit/psrpc/pkg/bus/redisbus"
+	"github.com/redis/go-redis/v9"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap/zapcore"
 	"go.uber.org/zap/zaptest/observer"
@@ -410,3 +418,50 @@ func TestLoggerCallDepth(t *testing.T) {
 		})
 	}
 }
+
+// chanWriteSyncer hands each write to the test, so a line logged from another goroutine is read
+// without racing on a shared buffer.
+type chanWriteSyncer chan []byte
+
+func (c chanWriteSyncer) Write(p []byte) (int, error) {
+	c <- bytes.Clone(p)
+	return len(p), nil
+}
+
+func (chanWriteSyncer) Sync() error { return nil }
+
+func TestSetLoggerRoutesPsrpcBusErrors(t *testing.T) {
+	t.Cleanup(func() {
+		defaultLogger = LogRLogger(discardLogger)
+		pkgLogger = LogRLogger(discardLogger)
+	})
+
+	lines := make(chanWriteSyncer, 1)
+	SetLogger(must.Get(NewZapLogger(&Config{}, jsonTee(lines))), "TEST")
+
+	dialErr := errors.New("redis is unreachable")
+	var dials atomic.Int32
+	rc := redis.NewClient(&redis.Options{
+		Dialer: func(context.Context, string, string) (net.Conn, error) {
+			if dials.Add(1) == 1 {
+				return nil, dialErr
+			}
+			// Park every reconnect after the first failure, so the bus logs once and then stays
+			// quiet instead of retrying into later tests.
+			select {}
+		},
+	})
+	redisbus.New(rc)
+
+	var line map[string]any
+	select {
+	case b := <-lines:
+		require.NoError(t, json.Unmarshal(b, &line))
+	case <-time.After(5 * time.Second):
+		t.Fatal("the psrpc bus error never reached the configured logger")
+	}
+	require.Equal(t, "error", line["level"])
+	require.Equal(t, "redis receive message failed", line["msg"])
+	require.Equal(t, "TEST.psrpc", line["logger"])
+	require.Contains(t, fmt.Sprint(line), dialErr.Error())
+}

logger/slog.go

@@ -32,7 +32,7 @@ func ToSlogHandler(log Logger) slog.Handler {
 	switch log := log.(type) {
 	case ZapLogger:
 		zlog := log.ToZap().Desugar()
-		return zapslog.NewHandler(zlog.Core())
+		return zapslog.NewHandler(zlog.Core(), zapslog.WithName(zlog.Name()))
 	case LogRLogger:
 		return logr.ToSlogHandler(log.toLogr())
 	}