From 0d27ae4179dfb9e944cf1e56b59f0bf90ee4f1dc Mon Sep 17 00:00:00 2001 From: Erik Hortsch Date: Thu, 20 Aug 2026 08:55:16 -0700 Subject: [PATCH 1/3] rpc: gate the claim skip process-wide for every server Wires psrpc's WithServerSkipClaim into WithServerObservability, which is the one seam every server constructor shares -- WithDefaultServerOptions calls it, and the constructors that take only a logger reach it too. The setting is process-wide because the claim is a transport policy rather than a per-service one, and it is read per request so callers may set it before or after their servers exist and revoke it without a redeploy. Co-Authored-By: Claude Opus 5 --- go.mod | 2 +- go.sum | 2 ++ rpc/typed_api.go | 23 +++++++++++++++++++++++ rpc/typed_api_test.go | 17 +++++++++++++++++ 4 files changed, 43 insertions(+), 1 deletion(-) diff --git a/go.mod b/go.mod index 3048626ca..416128399 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-0.20260819230101-cbf56a2f6872 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..484ed6c44 100644 --- a/go.sum +++ b/go.sum @@ -89,6 +89,8 @@ 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/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..31280222c 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,34 @@ func (p *ClientParams) Args() (psrpc.MessageBus, psrpc.ClientOption) { return p.Bus, psrpc.WithClientOptions(p.Options()...) } +var serverSkipClaim atomic.Pointer[func() bool] + +// SetServerSkipClaim gates the claim handshake for every server built through +// WithServerObservability, which is the only seam every server constructor +// shares -- several take no config to read the setting from. Process-wide +// because the claim is a transport policy, not a per-service one. Consulted per +// request, so this may be called before or after the servers exist, and the +// setting stays revocable at runtime. +func SetServerSkipClaim(enabled func() bool) { + serverSkipClaim.Store(&enabled) +} + +func serverSkipClaimEnabled() bool { + if enabled := serverSkipClaim.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{}), + // Rides along here rather than in WithDefaultServerOptions so it also + // reaches the servers that only take a logger. + psrpc.WithServerSkipClaim(serverSkipClaimEnabled), ) } diff --git a/rpc/typed_api_test.go b/rpc/typed_api_test.go index ffef3a94f..764b5a119 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() { serverSkipClaim.Store(nil) }) + + require.False(t, serverSkipClaimEnabled(), "unset must mean claim") + + var on atomic.Bool + SetServerSkipClaim(on.Load) + require.False(t, serverSkipClaimEnabled()) + + on.Store(true) + require.True(t, serverSkipClaimEnabled(), "must be read per call, not captured") + + on.Store(false) + require.False(t, serverSkipClaimEnabled(), "must stay revocable") +} From 0b5da630adb6fb4fbb554ac80ae8fc86c413be95 Mon Sep 17 00:00:00 2001 From: Erik Hortsch Date: Mon, 24 Aug 2026 09:10:27 -0700 Subject: [PATCH 2/3] pin psrpc v0.7.5, trim comments Co-Authored-By: Claude Opus 5 --- go.mod | 2 +- go.sum | 2 ++ rpc/typed_api.go | 11 +++-------- 3 files changed, 6 insertions(+), 9 deletions(-) diff --git a/go.mod b/go.mod index 416128399..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.5-0.20260819230101-cbf56a2f6872 + 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 484ed6c44..aec7f9fb8 100644 --- a/go.sum +++ b/go.sum @@ -91,6 +91,8 @@ 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 31280222c..4d3cdf5a2 100644 --- a/rpc/typed_api.go +++ b/rpc/typed_api.go @@ -100,12 +100,8 @@ func (p *ClientParams) Args() (psrpc.MessageBus, psrpc.ClientOption) { var serverSkipClaim atomic.Pointer[func() bool] -// SetServerSkipClaim gates the claim handshake for every server built through -// WithServerObservability, which is the only seam every server constructor -// shares -- several take no config to read the setting from. Process-wide -// because the claim is a transport policy, not a per-service one. Consulted per -// request, so this may be called before or after the servers exist, and the -// setting stays revocable at runtime. +// SetServerSkipClaim gates the claim skip for every server built through +// WithServerObservability. Process-wide; read per request, so revocable at runtime. func SetServerSkipClaim(enabled func() bool) { serverSkipClaim.Store(&enabled) } @@ -123,8 +119,7 @@ func WithServerObservability(logger logger.Logger) psrpc.ServerOption { psrpc.WithServerObserver(PSRPCMetricsObserver{}), WithServerLogger(logger), otelpsrpc.ServerOptions(otelpsrpc.Config{}), - // Rides along here rather than in WithDefaultServerOptions so it also - // reaches the servers that only take a logger. + // here rather than WithDefaultServerOptions so logger-only servers get it too psrpc.WithServerSkipClaim(serverSkipClaimEnabled), ) } From 7674583ae935a305cb7d0c5770187379b813ec23 Mon Sep 17 00:00:00 2001 From: Erik Hortsch Date: Mon, 24 Aug 2026 11:01:57 -0700 Subject: [PATCH 3/3] rename the skip-claim gate to carry the psrpc prefix Server and client mean too many things in this package for a bare SetServerSkipClaim to read clearly. Co-Authored-By: Claude Opus 5 --- rpc/typed_api.go | 14 +++++++------- rpc/typed_api_test.go | 12 ++++++------ 2 files changed, 13 insertions(+), 13 deletions(-) diff --git a/rpc/typed_api.go b/rpc/typed_api.go index 4d3cdf5a2..8a17819fa 100644 --- a/rpc/typed_api.go +++ b/rpc/typed_api.go @@ -98,16 +98,16 @@ func (p *ClientParams) Args() (psrpc.MessageBus, psrpc.ClientOption) { return p.Bus, psrpc.WithClientOptions(p.Options()...) } -var serverSkipClaim atomic.Pointer[func() bool] +var psrpcServerSkipClaim atomic.Pointer[func() bool] -// SetServerSkipClaim gates the claim skip for every server built through +// SetPSRPCServerSkipClaim gates the claim skip for every server built through // WithServerObservability. Process-wide; read per request, so revocable at runtime. -func SetServerSkipClaim(enabled func() bool) { - serverSkipClaim.Store(&enabled) +func SetPSRPCServerSkipClaim(enabled func() bool) { + psrpcServerSkipClaim.Store(&enabled) } -func serverSkipClaimEnabled() bool { - if enabled := serverSkipClaim.Load(); enabled != nil { +func psrpcServerSkipClaimEnabled() bool { + if enabled := psrpcServerSkipClaim.Load(); enabled != nil { return (*enabled)() } return false @@ -120,7 +120,7 @@ func WithServerObservability(logger logger.Logger) psrpc.ServerOption { WithServerLogger(logger), otelpsrpc.ServerOptions(otelpsrpc.Config{}), // here rather than WithDefaultServerOptions so logger-only servers get it too - psrpc.WithServerSkipClaim(serverSkipClaimEnabled), + psrpc.WithServerSkipClaim(psrpcServerSkipClaimEnabled), ) } diff --git a/rpc/typed_api_test.go b/rpc/typed_api_test.go index 764b5a119..fe27e8775 100644 --- a/rpc/typed_api_test.go +++ b/rpc/typed_api_test.go @@ -79,17 +79,17 @@ func TestPropagateRequestTimeout(t *testing.T) { } func TestServerSkipClaim(t *testing.T) { - t.Cleanup(func() { serverSkipClaim.Store(nil) }) + t.Cleanup(func() { psrpcServerSkipClaim.Store(nil) }) - require.False(t, serverSkipClaimEnabled(), "unset must mean claim") + require.False(t, psrpcServerSkipClaimEnabled(), "unset must mean claim") var on atomic.Bool - SetServerSkipClaim(on.Load) - require.False(t, serverSkipClaimEnabled()) + SetPSRPCServerSkipClaim(on.Load) + require.False(t, psrpcServerSkipClaimEnabled()) on.Store(true) - require.True(t, serverSkipClaimEnabled(), "must be read per call, not captured") + require.True(t, psrpcServerSkipClaimEnabled(), "must be read per call, not captured") on.Store(false) - require.False(t, serverSkipClaimEnabled(), "must stay revocable") + require.False(t, psrpcServerSkipClaimEnabled(), "must stay revocable") }