Add the ability of using prefixes in pubsub channel names

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

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) {

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)

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

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](