rpc: gate the claim skip process-wide for every server (#1729)

* 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 <noreply@anthropic.com>

* pin psrpc v0.7.5, trim comments

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

Author: erikhortsch

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

go.sum

@@ -87,6 +87,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=

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

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