Support key prefix for task queues

Author: cschleiden

backend/redis/queue.go

@@ -34,16 +34,21 @@ type KeyInfo struct {
 	SetKey    string
 }
 
-func newTaskQueue[T any](rdb redis.UniversalClient, tasktype string) (*taskQueue[T], error) {
+func newTaskQueue[T any](rdb redis.UniversalClient, keyPrefix string, tasktype string) (*taskQueue[T], error) {
+	// Ensure the key prefix ends with a colon
+	if keyPrefix != "" && keyPrefix[len(keyPrefix)-1] != ':' {
+		keyPrefix += ":"
+	}
+
 	tq := &taskQueue[T]{
 		tasktype:   tasktype,
-		setKey:     "task-set:" + tasktype,
-		streamKey:  "task-stream:" + tasktype,
+		setKey:     keyPrefix + "task-set:" + tasktype,
+		streamKey:  keyPrefix + "task-stream:" + tasktype,
 		groupName:  "task-workers",
 		workerName: uuid.NewString(),
 	}
 
-	// Pre-load script
+	// Pre-load scripts
 	cmds := map[string]*redis.StringCmd{
 		"createGroupCmd": createGroupCmd.Load(context.Background(), rdb),
 		"enqueueCmd":     enqueueCmd.Load(context.Background(), rdb),

backend/redis/queue_test.go

@@ -33,15 +33,15 @@ func Test_TaskQueue(t *testing.T) {
 		{
 			name: "Create queue",
 			f: func(t *testing.T) {
-				q, err := newTaskQueue[any](client, "test")
+				q, err := newTaskQueue[any](client, "", "test")
 				require.NoError(t, err)
 				require.NotNil(t, q)
 			},
 		},
 		{
 			name: "Simple enqueue/dequeue",
 			f: func(t *testing.T) {
-				q, err := newTaskQueue[any](client, "test")
+				q, err := newTaskQueue[any](client, "prefix", "test")
 				require.NoError(t, err)
 
 				ctx := context.Background()
@@ -60,7 +60,7 @@ func Test_TaskQueue(t *testing.T) {
 		{
 			name: "Guarantee uniqueness",
 			f: func(t *testing.T) {
-				q, err := newTaskQueue[any](client, "test")
+				q, err := newTaskQueue[any](client, "prefix", "test")
 				require.NoError(t, err)
 
 				ctx := context.Background()
@@ -101,7 +101,7 @@ func Test_TaskQueue(t *testing.T) {
 
 				ctx := context.Background()
 
-				q, err := newTaskQueue[foo](client, "test")
+				q, err := newTaskQueue[foo](client, "prefix", "test")
 				require.NoError(t, err)
 
 				_, err = client.Pipelined(ctx, func(p redis.Pipeliner) error {
@@ -123,7 +123,7 @@ func Test_TaskQueue(t *testing.T) {
 		{
 			name: "Simple enqueue/dequeue different worker",
 			f: func(t *testing.T) {
-				q, _ := newTaskQueue[any](client, "test")
+				q, _ := newTaskQueue[any](client, "prefix", "test")
 
 				ctx := context.Background()
 
@@ -132,7 +132,7 @@ func Test_TaskQueue(t *testing.T) {
 				})
 				require.NoError(t, err)
 
-				q2, _ := newTaskQueue[any](client, "test")
+				q2, _ := newTaskQueue[any](client, "prefix", "test")
 				require.NoError(t, err)
 
 				// Dequeue using second worker
@@ -145,8 +145,8 @@ func Test_TaskQueue(t *testing.T) {
 		{
 			name: "Complete removes task",
 			f: func(t *testing.T) {
-				q, _ := newTaskQueue[any](client, "test")
-				q2, _ := newTaskQueue[any](client, "test")
+				q, _ := newTaskQueue[any](client, "prefix", "test")
+				q2, _ := newTaskQueue[any](client, "prefix", "test")
 
 				ctx := context.Background()
 
@@ -177,7 +177,7 @@ func Test_TaskQueue(t *testing.T) {
 		{
 			name: "Recover task",
 			f: func(t *testing.T) {
-				q, _ := newTaskQueue[any](client, "test")
+				q, _ := newTaskQueue[any](client, "prefix", "test")
 
 				ctx := context.Background()
 
@@ -186,7 +186,7 @@ func Test_TaskQueue(t *testing.T) {
 				})
 				require.NoError(t, err)
 
-				q2, _ := newTaskQueue[any](client, "test")
+				q2, _ := newTaskQueue[any](client, "prefix", "test")
 				require.NoError(t, err)
 
 				task, err := q2.Dequeue(ctx, client, lockTimeout, blockTimeout)
@@ -206,7 +206,7 @@ func Test_TaskQueue(t *testing.T) {
 		{
 			name: "Extending task prevents recovering",
 			f: func(t *testing.T) {
-				q, _ := newTaskQueue[any](client, "test")
+				q, _ := newTaskQueue[any](client, "prefix", "test")
 
 				ctx := context.Background()
 
@@ -216,7 +216,7 @@ func Test_TaskQueue(t *testing.T) {
 				require.NoError(t, err)
 
 				// Create second worker (with different name)
-				q2, _ := newTaskQueue[any](client, "test")
+				q2, _ := newTaskQueue[any](client, "prefix", "test")
 				require.NoError(t, err)
 
 				task, err := q2.Dequeue(ctx, client, lockTimeout, blockTimeout)

backend/redis/redis.go

@@ -30,16 +30,6 @@ var (
 )
 
 func NewRedisBackend(client redis.UniversalClient, opts ...RedisBackendOption) (*redisBackend, error) {
-	workflowQueue, err := newTaskQueue[any](client, "workflows")
-	if err != nil {
-		return nil, fmt.Errorf("creating workflow task queue: %w", err)
-	}
-
-	activityQueue, err := newTaskQueue[activityData](client, "activities")
-	if err != nil {
-		return nil, fmt.Errorf("creating activity task queue: %w", err)
-	}
-
 	// Default options
 	options := &RedisOptions{
 		Options:      backend.ApplyOptions(),
@@ -50,6 +40,16 @@ func NewRedisBackend(client redis.UniversalClient, opts ...RedisBackendOption) (
 		opt(options)
 	}
 
+	workflowQueue, err := newTaskQueue[any](client, options.KeyPrefix, "workflows")
+	if err != nil {
+		return nil, fmt.Errorf("creating workflow task queue: %w", err)
+	}
+
+	activityQueue, err := newTaskQueue[activityData](client, options.KeyPrefix, "activities")
+	if err != nil {
+		return nil, fmt.Errorf("creating activity task queue: %w", err)
+	}
+
 	rb := &redisBackend{
 		rdb:     client,
 		options: options,