diff --git a/go.mod b/go.mod index 3048626ca..1c07aa907 100644 --- a/go.mod +++ b/go.mod @@ -16,7 +16,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/lithammer/shortuuid/v4 v4.2.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 - github.com/livekit/psrpc v0.7.3 + github.com/livekit/psrpc v0.7.5 github.com/mackerelio/go-osstat v0.2.8 github.com/maxbrunsfeld/counterfeiter/v6 v6.12.2 github.com/nyaruka/phonenumbers v1.8.1 diff --git a/go.sum b/go.sum index 4330d4a19..aec7f9fb8 100644 --- a/go.sum +++ b/go.sum @@ -89,6 +89,10 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/psrpc v0.7.3 h1:bekuZt/ZQzg8+/M8G6G5jq7bvV9fAKdPHSOZeTwrIIc= github.com/livekit/psrpc v0.7.3/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= +github.com/livekit/psrpc v0.7.5-0.20260819230101-cbf56a2f6872 h1:T4+LTChYiNKWkK2yeW0FXB2CNpEXxiHEV75Xhh8lDUI= +github.com/livekit/psrpc v0.7.5-0.20260819230101-cbf56a2f6872/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= +github.com/livekit/psrpc v0.7.5 h1:WxfJIQ41X1b+48A1uzc8Gy9FhYEMxBJNCpCYqPdO/Ds= +github.com/livekit/psrpc v0.7.5/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= github.com/mackerelio/go-osstat v0.2.8 h1:I2duicTaCGWoM53XwAwA9OIe1inu0xnVs8/pqOWWVr4= github.com/mackerelio/go-osstat v0.2.8/go.mod h1:SyS3XxKdoSKJnTGTkN5Yrh6VUQVuAURACfE6y+2DN4k= github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= diff --git a/rpc/typed_api.go b/rpc/typed_api.go index 49189d928..8a17819fa 100644 --- a/rpc/typed_api.go +++ b/rpc/typed_api.go @@ -17,6 +17,7 @@ package rpc import ( "context" "fmt" + "sync/atomic" "time" "github.com/livekit/psrpc" @@ -97,12 +98,29 @@ func (p *ClientParams) Args() (psrpc.MessageBus, psrpc.ClientOption) { return p.Bus, psrpc.WithClientOptions(p.Options()...) } +var psrpcServerSkipClaim atomic.Pointer[func() bool] + +// SetPSRPCServerSkipClaim gates the claim skip for every server built through +// WithServerObservability. Process-wide; read per request, so revocable at runtime. +func SetPSRPCServerSkipClaim(enabled func() bool) { + psrpcServerSkipClaim.Store(&enabled) +} + +func psrpcServerSkipClaimEnabled() bool { + if enabled := psrpcServerSkipClaim.Load(); enabled != nil { + return (*enabled)() + } + return false +} + func WithServerObservability(logger logger.Logger) psrpc.ServerOption { return psrpc.WithServerOptions( middleware.WithServerMetrics(PSRPCMetricsObserver{}), psrpc.WithServerObserver(PSRPCMetricsObserver{}), WithServerLogger(logger), otelpsrpc.ServerOptions(otelpsrpc.Config{}), + // here rather than WithDefaultServerOptions so logger-only servers get it too + psrpc.WithServerSkipClaim(psrpcServerSkipClaimEnabled), ) } diff --git a/rpc/typed_api_test.go b/rpc/typed_api_test.go index ffef3a94f..fe27e8775 100644 --- a/rpc/typed_api_test.go +++ b/rpc/typed_api_test.go @@ -6,6 +6,7 @@ import ( reflect "reflect" "runtime" "slices" + "sync/atomic" "testing" "time" @@ -76,3 +77,19 @@ func TestPropagateRequestTimeout(t *testing.T) { WithPropagateRequestTimeout(ctx)(&ro) require.InEpsilon(t, 42*time.Second, ro.Timeout, 0.01) } + +func TestServerSkipClaim(t *testing.T) { + t.Cleanup(func() { psrpcServerSkipClaim.Store(nil) }) + + require.False(t, psrpcServerSkipClaimEnabled(), "unset must mean claim") + + var on atomic.Bool + SetPSRPCServerSkipClaim(on.Load) + require.False(t, psrpcServerSkipClaimEnabled()) + + on.Store(true) + require.True(t, psrpcServerSkipClaimEnabled(), "must be read per call, not captured") + + on.Store(false) + require.False(t, psrpcServerSkipClaimEnabled(), "must stay revocable") +}