fix: hash-tag rsmq queue keys so retries work on redis cluster

cmd/outpost/migrate.go

@@ -69,6 +69,19 @@ func newMigrateCommand() *cli.Command {
 				Usage:  "Verify that migrations were applied correctly",
 				Action: runMigrateVerify,
 			},
+			{
+				Name:      "cleanup",
+				Usage:     "Remove data an applied Redis migration left behind (run once the previous version is fully rolled out)",
+				ArgsUsage: "<migration>",
+				Flags: []cli.Flag{
+					&cli.BoolFlag{
+						Name:    "yes",
+						Aliases: []string{"y"},
+						Usage:   "Skip confirmation prompt",
+					},
+				},
+				Action: runMigrateCleanup,
+			},
 			{
 				Name:  "unlock",
 				Usage: "Force clear the Redis migration lock (use with caution)",
@@ -225,6 +238,53 @@ func runMigrateVerify(ctx context.Context, c *cli.Command) error {
 	})
 }
 
+func runMigrateCleanup(ctx context.Context, c *cli.Command) error {
+	return withCoordinator(ctx, c, func(coord *coordinator.Coordinator) error {
+		name := c.Args().First()
+		if name == "" {
+			plans, err := coord.PlanCleanup(ctx)
+			if err != nil {
+				return err
+			}
+			printCleanupCandidates(os.Stdout, plans)
+			return nil
+		}
+
+		plan, err := coord.PlanCleanupFor(ctx, name)
+		if err != nil {
+			return err
+		}
+		if plan.NotApplicableReason != "" {
+			fmt.Fprintf(os.Stdout, "redis/%s is not applicable (%s); nothing to clean up.\n", plan.Name, plan.NotApplicableReason)
+			return nil
+		}
+		if plan.Items == 0 {
+			fmt.Fprintf(os.Stdout, "Nothing to clean up for redis/%s.\n", plan.Name)
+			return nil
+		}
+
+		fmt.Fprintf(os.Stdout, "redis/%s: %s\n  keys to clean up: %d\n", plan.Name, plan.Description, plan.Items)
+		if !c.Bool("yes") {
+			fmt.Fprint(os.Stdout, "\nThis deletes old data and cannot be undone. Continue? [y/N]: ")
+			var response string
+			if _, err := fmt.Fscanln(os.Stdin, &response); err != nil {
+				fmt.Fprintln(os.Stdout, "Cancelled.")
+				return nil
+			}
+			if response != "y" && response != "Y" && response != "yes" {
+				fmt.Fprintln(os.Stdout, "Cancelled.")
+				return nil
+			}
+		}
+
+		if err := coord.Cleanup(ctx, plan.Name); err != nil {
+			return err
+		}
+		fmt.Fprintf(os.Stdout, "\nCleanup of redis/%s complete.\n", plan.Name)
+		return nil
+	})
+}
+
 func runMigrateUnlock(ctx context.Context, c *cli.Command) error {
 	return withCoordinator(ctx, c, func(coord *coordinator.Coordinator) error {
 		if !c.Bool("yes") {

cmd/outpost/migrate_output.go

@@ -126,3 +126,18 @@ func printVerificationReport(w io.Writer, report *coordinator.VerificationReport
 		fmt.Fprintln(w, "Verification reported issues.")
 	}
 }
+
+// printCleanupCandidates renders `outpost migrate cleanup` without a
+// migration name: what each applied migration would clean up.
+func printCleanupCandidates(w io.Writer, plans []coordinator.RedisCleanupPlan) {
+	if len(plans) == 0 {
+		fmt.Fprintln(w, "Nothing to clean up.")
+		return
+	}
+	fmt.Fprintln(w, "Applied migrations with data to clean up:")
+	for _, p := range plans {
+		fmt.Fprintf(w, "  - redis/%s: %s\n      keys to clean up: %d\n", p.Name, p.Description, p.Items)
+	}
+	fmt.Fprintln(w)
+	fmt.Fprintln(w, "Run 'outpost migrate cleanup <migration>' to clean up one of them.")
+}

cmd/redis-debug/main.go

@@ -145,6 +145,11 @@ type RedisClientInterface interface {
 func testRedisOperations(client RedisClientInterface) {
 	// Check specific RSMQ keys
 	rsmqKeys := []string{
+		"rsmq:{deliverymq-retry}:Q",
+		"rsmq:{deliverymq-retry}",
+		"rsmq:{deliverymq-retry-dlq}:Q",
+		"rsmq:{deliverymq-retry-dlq}",
+		// Untagged layout from earlier versions; `outpost migrate` moves it.
 		"rsmq:deliverymq-retry:Q",
 		"rsmq:deliverymq-retry",
 		"rsmq:deliverymq-retry-dlq:Q",

docs/content/self-hosting/data-model.mdoc

@@ -203,10 +203,10 @@ Scheduled retries are stored in an [RSMQ](https://github.com/smrchy/rsmq) queue
 | Key | Type | Contents |
 |-----|------|----------|
 | `rsmq:QUEUES` | Set | Names of the queues Outpost has created. |
-| `rsmq:deliverymq-retry` | Sorted set | Pending retry IDs, scored by the time each becomes due. |
-| `rsmq:deliverymq-retry:Q` | Hash | Queue settings, plus each retry's body and receive counters. |
-| `rsmq:deliverymq-retry-dlq` | Sorted set | Retries that failed to be processed after 5 receives. |
-| `rsmq:deliverymq-retry-dlq:Q` | Hash | Bodies for dead-lettered retries. |
+| `rsmq:{deliverymq-retry}` | Sorted set | Pending retry IDs, scored by the time each becomes due. |
+| `rsmq:{deliverymq-retry}:Q` | Hash | Queue settings, plus each retry's body and receive counters. |
+| `rsmq:{deliverymq-retry-dlq}` | Sorted set | Retries that failed to be processed after 5 receives. |
+| `rsmq:{deliverymq-retry-dlq}:Q` | Hash | Bodies for dead-lettered retries. |
 
 Each retry's ID is derived from its event ID and destination ID, so there is at most one pending retry per event and destination. Scheduling a new retry, or making a manual retry, replaces or cancels the pending one.
 
@@ -240,7 +240,7 @@ These keys are short-lived and self-expiring, and safe to lose. Losing an idempo
 Following one event through the system shows how the stores relate:
 
 1. **Publish.** The API service, or the publish queue consumer, claims `idempotency:publishmq:<event_id>`. It reads `tenant:{<tenant_id>}:destinations` to find the destinations that match the event's topic and filters, then enqueues one delivery task per destination on the delivery queue.
-2. **Deliver.** The delivery service claims `idempotency:deliverymq:<event_id>:<destination_id>:<attempt>` and reads `tenant:{<tenant_id>}:destination:<destination_id>` for the destination's config and decrypted credentials. It makes the attempt and sends the result to the log queue. On a retryable failure, it also schedules a retry in `rsmq:deliverymq-retry`.
+2. **Deliver.** The delivery service claims `idempotency:deliverymq:<event_id>:<destination_id>:<attempt>` and reads `tenant:{<tenant_id>}:destination:<destination_id>` for the destination's config and decrypted credentials. It makes the attempt and sends the result to the log queue. On a retryable failure, it also schedules a retry in `rsmq:{deliverymq-retry}`.
 3. **Log.** The log service writes the event to `events`, if it isn't there yet, and the attempt to `attempts`, in batches. For failed attempts, it updates `alert:<tenant_id>:<destination_id>:cf` and may emit operator events or disable the destination. A successful attempt resets the count.
 4. **Retry.** When a retry is due, the API service, which runs the retry scheduler, reads the event and the latest attempt from `attempts`, then enqueues a new delivery task with the next attempt number, which returns to step 2.
 

docs/content/self-hosting/guides/migration.mdoc

@@ -120,6 +120,20 @@ This performs spot-checks on a sample of migrated data to ensure integrity. Addi
 - Verify critical workflows
 - Monitor for any errors
 
+#### Step 4: Clean Up (when a migration has a cleanup step)
+
+Some Redis migrations leave old data in place so instances still running the previous version keep working during a rolling upgrade. Once no instance of the previous version is left, remove it:
+
+```bash
+# List applied migrations with data to clean up
+docker run --rm hookdeck/outpost migrate cleanup
+
+# Clean up one migration
+docker run --rm -it hookdeck/outpost migrate cleanup 004_rsmq_hash_tags
+```
+
+Cleanup deletes data and cannot be undone. Add `--yes` to skip the confirmation. Each migration's release notes say whether it needs a cleanup and when to run it.
+
 
 ## Safety Features
 

docs/content/self-hosting/guides/troubleshooting-redis.mdoc

@@ -128,6 +128,10 @@ go build -o redis-debug cmd/redis-debug/main.go
 ✅ TIME: 2025-08-21 22:15:30.123456 +0100 BST
 
 === RSMQ Key Analysis ===
+ℹ️  rsmq:{deliverymq-retry}:Q: Does not exist
+ℹ️  rsmq:{deliverymq-retry}: Does not exist
+ℹ️  rsmq:{deliverymq-retry-dlq}:Q: Does not exist
+ℹ️  rsmq:{deliverymq-retry-dlq}: Does not exist
 ℹ️  rsmq:deliverymq-retry:Q: Does not exist
 ℹ️  rsmq:deliverymq-retry: Does not exist
 ℹ️  rsmq:deliverymq-retry-dlq:Q: Does not exist

internal/deliverymq/retry.go

@@ -55,20 +55,19 @@ func NewRetryScheduler(deliverymq *DeliveryMQ, redisConfig *redis.RedisConfig, d
 
 	adapter := rsmq.NewRedisAdapter(redisClient)
 
-	// Construct RSMQ namespace with deployment prefix if provided
-	// This creates keys like: dp_001:rsmq:QUEUES, dp_001:rsmq:deliverymq-retry:Q
-	// Without deployment ID: rsmq:QUEUES, rsmq:deliverymq-retry:Q
+	// Construct RSMQ namespace with deployment prefix if provided. The
+	// deployment prefix also goes inside the hash tag so deployments sharing a
+	// Redis Cluster spread across slots.
+	// This creates keys like: dp_001:rsmq:QUEUES, dp_001:rsmq:{dp_001:deliverymq-retry}:Q
+	// Without deployment ID: rsmq:QUEUES, rsmq:{deliverymq-retry}:Q
 	namespace := "rsmq"
+	tagPrefix := ""
 	if deploymentID != "" {
 		namespace = fmt.Sprintf("%s:rsmq", deploymentID)
+		tagPrefix = deploymentID + ":"
 	}
 
-	var rsmqClient *rsmq.RedisSMQ
-	if logger != nil {
-		rsmqClient = rsmq.NewRedisSMQ(adapter, namespace, logger)
-	} else {
-		rsmqClient = rsmq.NewRedisSMQ(adapter, namespace)
-	}
+	rsmqClient := rsmq.NewRedisSMQ(adapter, namespace, rsmq.WithLogger(logger), rsmq.WithHashTagPrefix(tagPrefix))
 
 	exec := func(ctx context.Context, msg string) error {
 		retryTask := RetryTask{}

internal/deliverymq/retry_migration_test.go

@@ -0,0 +1,145 @@
+package deliverymq_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/hookdeck/outpost/internal/deliverymq"
+	"github.com/hookdeck/outpost/internal/idgen"
+	"github.com/hookdeck/outpost/internal/migrator/migratorredis"
+	migration_004 "github.com/hookdeck/outpost/internal/migrator/migratorredis/004_rsmq_hash_tags"
+	"github.com/hookdeck/outpost/internal/models"
+	"github.com/hookdeck/outpost/internal/mqs"
+	"github.com/hookdeck/outpost/internal/redis"
+	"github.com/hookdeck/outpost/internal/rsmq"
+	"github.com/hookdeck/outpost/internal/scheduler"
+	"github.com/hookdeck/outpost/internal/util/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// legacyRetryQueue schedules retries the way versions before hash-tagged
+// rsmq keys did, and reports how many remain in that queue.
+type legacyRetryQueue struct {
+	scheduler scheduler.Scheduler
+	client    redis.Client
+	ns        string
+}
+
+func newLegacyRetryQueue(t *testing.T, ctx context.Context, cfg *redis.RedisConfig, ns string) *legacyRetryQueue {
+	client, err := redis.New(ctx, cfg)
+	require.NoError(t, err)
+	t.Cleanup(func() { client.Close() })
+	s := scheduler.New("deliverymq-retry", rsmq.NewRedisSMQ(rsmq.NewRedisAdapter(client), ns, rsmq.WithUntaggedKeys()), nil)
+	require.NoError(t, s.Init(ctx))
+	return &legacyRetryQueue{scheduler: s, client: client, ns: ns}
+}
+
+func (q *legacyRetryQueue) schedule(t *testing.T, ctx context.Context, task deliverymq.RetryTask, delay time.Duration) {
+	msg, err := task.ToString()
+	require.NoError(t, err)
+	retryID := models.RetryID(task.EventID, task.DestinationID)
+	require.NoError(t, q.scheduler.Schedule(ctx, msg, delay, scheduler.WithTaskID(retryID)))
+}
+
+func (q *legacyRetryQueue) len(t *testing.T, ctx context.Context) int64 {
+	n, err := q.client.ZCard(ctx, q.ns+":deliverymq-retry").Result()
+	require.NoError(t, err)
+	return n
+}
+
+func TestRetryScheduler_MigratedLegacyQueue(t *testing.T) {
+	tests := []struct {
+		name         string
+		deploymentID string
+		legacyNs     string
+		currentKey   string
+	}{
+		{name: "without deployment id", deploymentID: "", legacyNs: "rsmq", currentKey: "rsmq:{deliverymq-retry}"},
+		{name: "with deployment id", deploymentID: "dp_legacy", legacyNs: "dp_legacy:rsmq", currentKey: "dp_legacy:rsmq:{dp_legacy:deliverymq-retry}"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			testRetrySchedulerMigratedLegacyQueue(t, tt.deploymentID, tt.legacyNs, tt.currentKey)
+		})
+	}
+}
+
+func testRetrySchedulerMigratedLegacyQueue(t *testing.T, deploymentID, legacyNs, currentKey string) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	tenant := models.Tenant{ID: idgen.String()}
+	destination := testutil.DestinationFactory.Any(
+		testutil.DestinationFactory.WithType("webhook"),
+		testutil.DestinationFactory.WithTenantID(tenant.ID),
+	)
+	event := testutil.EventFactory.Any(
+		testutil.EventFactory.WithTenantID(tenant.ID),
+		testutil.EventFactory.WithDestinationID(destination.ID),
+		testutil.EventFactory.WithEligibleForRetry(true),
+	)
+	eventGetter := newMockEventGetter()
+	eventGetter.addRecord(&models.Attempt{
+		ID:            idgen.Attempt(),
+		EventID:       event.ID,
+		DestinationID: destination.ID,
+		AttemptNumber: 1,
+	}, &event)
+
+	redisConfig := testutil.CreateTestRedisConfig(t)
+	legacy := newLegacyRetryQueue(t, ctx, redisConfig, legacyNs)
+	retryTask := deliverymq.RetryTask{EventID: event.ID, TenantID: tenant.ID, DestinationID: destination.ID}
+
+	// A retry scheduled by the previous version, then the migration.
+	legacy.schedule(t, ctx, retryTask, 0)
+	require.Equal(t, int64(1), legacy.len(t, ctx))
+	m := migration_004.New(legacy.client, migratorredis.NewLoggerAdapter(testutil.CreateTestLogger(t), false), deploymentID)
+	plan, err := m.Plan(ctx)
+	require.NoError(t, err)
+	_, err = m.Apply(ctx, plan)
+	require.NoError(t, err)
+	require.Equal(t, int64(0), legacy.len(t, ctx))
+
+	mqConfig := &mqs.QueueConfig{InMemory: &mqs.InMemoryConfig{Name: testutil.RandomString(5)}}
+	deliveryMQ := deliverymq.New(deliverymq.WithQueue(mqConfig))
+	cleanup, err := deliveryMQ.Init(ctx)
+	require.NoError(t, err)
+	defer cleanup()
+
+	retryScheduler, err := deliverymq.NewRetryScheduler(deliveryMQ, redisConfig, deploymentID, 10*time.Millisecond, testutil.CreateTestLogger(t), eventGetter)
+	require.NoError(t, err)
+	require.NoError(t, retryScheduler.Init(ctx))
+	defer retryScheduler.Shutdown()
+
+	subscription, err := deliveryMQ.Subscribe(ctx)
+	require.NoError(t, err)
+	defer subscription.Shutdown(ctx)
+	go retryScheduler.Monitor(ctx)
+
+	msg, err := subscription.Receive(ctx)
+	require.NoError(t, err)
+	msg.Ack()
+	var task models.DeliveryTask
+	require.NoError(t, task.FromMessage(msg))
+	assert.Equal(t, event.ID, task.Event.ID)
+	assert.Equal(t, destination.ID, task.DestinationID)
+	assert.Equal(t, 2, task.Attempt)
+
+	require.Eventually(t, func() bool {
+		n, err := legacy.client.ZCard(ctx, currentKey).Result()
+		return err == nil && n == 0
+	}, 2*time.Second, 10*time.Millisecond, "retry should be deleted after it fires")
+
+	t.Run("new retries go to the hash-tagged queue", func(t *testing.T) {
+		msg, err := retryTask.ToString()
+		require.NoError(t, err)
+		retryID := models.RetryID(event.ID, destination.ID)
+		require.NoError(t, retryScheduler.Schedule(ctx, msg, time.Hour, scheduler.WithTaskID(retryID)))
+		n, err := legacy.client.ZCard(ctx, currentKey).Result()
+		require.NoError(t, err)
+		assert.Equal(t, int64(1), n)
+		assert.Equal(t, int64(0), legacy.len(t, ctx))
+	})
+}