From bcd7600d83e159d5bfb2f3571120b3cb9f7b643d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?L=C3=A1szl=C3=B3=20K=C3=A1rolyi?= Date: Sat, 15 Aug 2026 18:55:23 +0200 Subject: [PATCH] bus: add configurable channel prefix for redis message bus Multiple psrpc deployments sharing one redis instance collide on pubsub channel names (redis DB index only scopes the keyspace, not PUBLISH/SUBSCRIBE). WithChannelPrefix lets callers namespace every channel the redis bus uses. Also widens TestRPC's fixed RPC-readiness sleeps to reduce unrelated CI flakiness (a fan-out request occasionally firing before all 3 servers finished registering). --- bus.go | 12 ++++++-- internal/bus/bus_redis.go | 44 ++++++++++++++++++++++----- internal/bus/bus_redis_test.go | 55 ++++++++++++++++++++++++++++++++++ internal/bus/bustest/redis.go | 16 ++++++++++ internal/test/psrpc_test.go | 6 ++-- 5 files changed, 120 insertions(+), 13 deletions(-) diff --git a/bus.go b/bus.go index f09660a..8905d30 100644 --- a/bus.go +++ b/bus.go @@ -23,6 +23,7 @@ import ( type Channel = bus.Channel type MessageBus bus.MessageBus +type RedisOption = bus.RedisOption func NewLocalMessageBus() MessageBus { return bus.NewLocalMessageBus() @@ -32,6 +33,13 @@ func NewNatsMessageBus(nc *nats.Conn) MessageBus { return bus.NewNatsMessageBus(nc) } -func NewRedisMessageBus(rc redis.UniversalClient) MessageBus { - return bus.NewRedisMessageBus(rc) +func NewRedisMessageBus(rc redis.UniversalClient, opts ...RedisOption) MessageBus { + return bus.NewRedisMessageBus(rc, opts...) +} + +// WithChannelPrefix prepends prefix to every redis pubsub channel name +// used by the bus, so multiple psrpc deployments can share one redis +// instance without colliding on channel names. +func WithChannelPrefix(prefix string) RedisOption { + return bus.WithChannelPrefix(prefix) } diff --git a/internal/bus/bus_redis.go b/internal/bus/bus_redis.go index 7d770e2..09c5077 100644 --- a/internal/bus/bus_redis.go +++ b/internal/bus/bus_redis.go @@ -42,9 +42,10 @@ const ( ) type redisMessageBus struct { - rc redis.UniversalClient - ctx context.Context - ps *redis.PubSub + rc redis.UniversalClient + ctx context.Context + ps *redis.PubSub + prefix string mu sync.Mutex subs map[string]*redisSubList @@ -58,12 +59,34 @@ type redisMessageBus struct { publishQueues [publishBuckets]*redisPublishQueue } -func NewRedisMessageBus(rc redis.UniversalClient) MessageBus { +// RedisOption configures optional behavior of the redis MessageBus. +type RedisOption func(*redisOpts) + +type redisOpts struct { + channelPrefix string +} + +// WithChannelPrefix prepends prefix to every redis pubsub channel name +// used by the bus, so multiple psrpc deployments can share one redis +// instance without colliding on channel names. +func WithChannelPrefix(prefix string) RedisOption { + return func(o *redisOpts) { + o.channelPrefix = prefix + } +} + +func NewRedisMessageBus(rc redis.UniversalClient, opts ...RedisOption) MessageBus { + o := &redisOpts{} + for _, opt := range opts { + opt(o) + } + ctx := context.Background() r := &redisMessageBus{ rc: rc, ctx: ctx, ps: rc.Subscribe(ctx), + prefix: o.channelPrefix, subs: map[string]*redisSubList{}, queues: map[string]*redisSubList{}, @@ -80,23 +103,28 @@ func NewRedisMessageBus(rc redis.UniversalClient) MessageBus { return r } +func (r *redisMessageBus) chanName(channel Channel) string { + return r.prefix + channel.Legacy +} + func (r *redisMessageBus) Publish(_ context.Context, channel Channel, msg proto.Message) error { b, err := serialize(msg, "") if err != nil { return err } - bucket := xxh3.HashString(channel.Legacy) % publishBuckets - r.publishQueues[bucket].Enqueue(channel.Legacy, b) + name := r.chanName(channel) + bucket := xxh3.HashString(name) % publishBuckets + r.publishQueues[bucket].Enqueue(name, b) return nil } func (r *redisMessageBus) Subscribe(ctx context.Context, channel Channel, size int) (Reader, error) { - return r.subscribe(ctx, channel.Legacy, size, r.subs, false) + return r.subscribe(ctx, r.chanName(channel), size, r.subs, false) } func (r *redisMessageBus) SubscribeQueue(ctx context.Context, channel Channel, size int) (Reader, error) { - return r.subscribe(ctx, channel.Legacy, size, r.queues, true) + return r.subscribe(ctx, r.chanName(channel), size, r.queues, true) } func (r *redisMessageBus) subscribe(ctx context.Context, channel string, size int, subLists map[string]*redisSubList, queue bool) (Reader, error) { diff --git a/internal/bus/bus_redis_test.go b/internal/bus/bus_redis_test.go index a7a6d9a..8550b47 100644 --- a/internal/bus/bus_redis_test.go +++ b/internal/bus/bus_redis_test.go @@ -21,6 +21,7 @@ import ( "time" "unsafe" + "github.com/redis/go-redis/v9" "github.com/stretchr/testify/require" "go.uber.org/atomic" "google.golang.org/protobuf/types/known/wrapperspb" @@ -29,6 +30,13 @@ import ( "github.com/livekit/psrpc/internal/bus/bustest" ) +// redisPrefixServer exposes the extra bustest.redisServer methods needed to +// test WithChannelPrefix, without widening the shared bustest.Server interface. +type redisPrefixServer interface { + Addr() string + ConnectWithOptions(t testing.TB, opts ...bus.RedisOption) bus.MessageBus +} + func redisTestChannel(channel string) bus.Channel { return bus.Channel{Legacy: channel} } @@ -93,6 +101,53 @@ func TestRedisMessageBus(t *testing.T) { require.EqualValues(t, 1, n.Load()) }) + t.Run("channel prefix is applied on the wire and isolates subscribers", func(t *testing.T) { + rs := srv.(redisPrefixServer) + + bA1 := rs.ConnectWithOptions(t, bus.WithChannelPrefix("tenantA:")) + bA2 := rs.ConnectWithOptions(t, bus.WithChannelPrefix("tenantA:")) + bB := rs.ConnectWithOptions(t, bus.WithChannelPrefix("tenantB:")) + + rA, err := bA2.Subscribe(context.Background(), redisTestChannel("test"), 100) + require.NoError(t, err) + rB, err := bB.Subscribe(context.Background(), redisTestChannel("test"), 100) + require.NoError(t, err) + + time.Sleep(100 * time.Millisecond) + + // the actual redis channel names must carry each bus's prefix; the + // shared test server may have other channels live from sibling + // subtests, so check containment rather than the full set. + rc := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{rs.Addr()}}) + defer rc.Close() + channels, err := rc.PubSubChannels(context.Background(), "*").Result() + require.NoError(t, err) + require.Subset(t, channels, []string{"tenantA:test", "tenantB:test"}) + + src := wrapperspb.String("hello") + require.NoError(t, bA1.Publish(context.Background(), redisTestChannel("test"), src)) + + // same prefix: message is received + b, ok := bus.RawRead(rA) + require.True(t, ok) + dst, err := bus.Deserialize(b) + require.NoError(t, err) + require.Equal(t, src.Value, dst.(*wrapperspb.StringValue).Value) + + // different prefix: message must not leak across the namespace + received := make(chan struct{}) + go func() { + if _, ok := bus.RawRead(rB); ok { + close(received) + } + }() + select { + case <-received: + t.Fatal("subscriber with a different channel prefix received a message published under another prefix") + case <-time.After(500 * time.Millisecond): + } + }) + t.Run("closed subscriptions are unreadable", func(t *testing.T) { b0 := srv.Connect(t) b1 := srv.Connect(t) diff --git a/internal/bus/bustest/redis.go b/internal/bus/bustest/redis.go index 9d7af19..111ab7a 100644 --- a/internal/bus/bustest/redis.go +++ b/internal/bus/bustest/redis.go @@ -78,3 +78,19 @@ func (s *redisServer) Connect(t testing.TB) bus.MessageBus { } return bus.NewRedisMessageBus(rc) } + +// Addr returns the redis server's host:port, for tests that need to talk +// to it directly (e.g. to inspect the raw pubsub channels in use). +func (s *redisServer) Addr() string { + return s.addr +} + +// ConnectWithOptions is like Connect, but forwards opts to +// bus.NewRedisMessageBus. +func (s *redisServer) ConnectWithOptions(t testing.TB, opts ...bus.RedisOption) bus.MessageBus { + rc, err := s.connect() + if err != nil { + t.Fatal(err) + } + return bus.NewRedisMessageBus(rc, opts...) +} diff --git a/internal/test/psrpc_test.go b/internal/test/psrpc_test.go index ed777ed..aa822a8 100644 --- a/internal/test/psrpc_test.go +++ b/internal/test/psrpc_test.go @@ -93,7 +93,7 @@ func testRPC(t *testing.T, bus func(t testing.TB) bus.MessageBus) { require.NoError(t, err) err = server.RegisterHandler[*internal.Request, *internal.Response](serverB, rpc, nil, addOne, nil) require.NoError(t, err) - time.Sleep(time.Second) + time.Sleep(2 * time.Second) ctx := context.Background() requestID := rand.NewRequestID() @@ -116,7 +116,7 @@ func testRPC(t *testing.T, bus func(t testing.TB) bus.MessageBus) { require.NoError(t, err) err = server.RegisterHandler[*internal.Request, *internal.Response](serverC, multiRpc, nil, returnError, nil) require.NoError(t, err) - time.Sleep(time.Second) + time.Sleep(2 * time.Second) requestID = rand.NewRequestID() resChan, err := client.RequestMulti[*internal.Response]( @@ -183,7 +183,7 @@ func testStream(t *testing.T, bus func(t testing.TB) bus.MessageBus) { err = server.RegisterStreamHandler[*internal.Response, *internal.Response](serverA, rpc, nil, handlePing, nil) require.NoError(t, err) - time.Sleep(time.Second) + time.Sleep(2 * time.Second) ctx := context.Background() stream, err := client.OpenStream[*internal.Response, *internal.Response](