From 0d27ae4179dfb9e944cf1e56b59f0bf90ee4f1dc Mon Sep 17 00:00:00 2001 From: Erik Hortsch Date: Thu, 20 Aug 2026 08:55:16 -0700 Subject: [PATCH] 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") +}