rpc: gate the claim skip process-wide for every server

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

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=

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),
 	)
 }
 

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")
+}