diff --git a/cmd/outpost/migrate.go b/cmd/outpost/migrate.go
index e41e09f98..0b80345f6 100644
--- a/cmd/outpost/migrate.go
+++ b/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") {
diff --git a/cmd/outpost/migrate_output.go b/cmd/outpost/migrate_output.go
index dbb4779d5..64b34e7ab 100644
--- a/cmd/outpost/migrate_output.go
+++ b/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.")
+}
diff --git a/cmd/redis-debug/main.go b/cmd/redis-debug/main.go
index 82d8b1581..83fc7a987 100644
--- a/cmd/redis-debug/main.go
+++ b/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",
diff --git a/docs/content/self-hosting/data-model.mdoc b/docs/content/self-hosting/data-model.mdoc
index 892628b3b..b905fba6b 100644
--- a/docs/content/self-hosting/data-model.mdoc
+++ b/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.
 
diff --git a/docs/content/self-hosting/guides/migration.mdoc b/docs/content/self-hosting/guides/migration.mdoc
index e85b02a53..4448767bf 100644
--- a/docs/content/self-hosting/guides/migration.mdoc
+++ b/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
 
diff --git a/docs/content/self-hosting/guides/troubleshooting-redis.mdoc b/docs/content/self-hosting/guides/troubleshooting-redis.mdoc
index 9a3845a5f..bc5d308a6 100644
--- a/docs/content/self-hosting/guides/troubleshooting-redis.mdoc
+++ b/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
diff --git a/internal/deliverymq/retry.go b/internal/deliverymq/retry.go
index e43bd38a8..ed1ea5814 100644
--- a/internal/deliverymq/retry.go
+++ b/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{}
diff --git a/internal/deliverymq/retry_migration_test.go b/internal/deliverymq/retry_migration_test.go
new file mode 100644
index 000000000..d1775c2a3
--- /dev/null
+++ b/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))
+	})
+}
diff --git a/internal/migrator/coordinator/coordinator.go b/internal/migrator/coordinator/coordinator.go
index 5b44b9861..32fc9f5c6 100644
--- a/internal/migrator/coordinator/coordinator.go
+++ b/internal/migrator/coordinator/coordinator.go
@@ -2,8 +2,10 @@ package coordinator
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sort"
+	"strings"
 	"time"
 
 	"github.com/hookdeck/outpost/internal/logging"
@@ -23,6 +25,14 @@ const (
 	redisLockTTL = time.Hour
 )
 
+var (
+	// ErrUnknownMigration is returned for a migration name that doesn't exist.
+	ErrUnknownMigration = errors.New("unknown migration")
+	// ErrNotApplied is returned when cleaning up a migration that hasn't been
+	// applied yet.
+	ErrNotApplied = errors.New("migration not applied yet; run 'outpost migrate apply' first")
+)
+
 // Coordinator unifies the SQL and Redis migration subsystems behind a
 // single API. It is intentionally safe to use from both a CLI and from
 // app startup checks.
@@ -361,6 +371,96 @@ func (c *Coordinator) Verify(ctx context.Context) (*VerificationReport, error) {
 	return report, nil
 }
 
+// PlanCleanup lists the applied Redis migrations that have old data left
+// to clean up.
+func (c *Coordinator) PlanCleanup(ctx context.Context) ([]RedisCleanupPlan, error) {
+	if c.redisClient == nil {
+		return nil, nil
+	}
+	var out []RedisCleanupPlan
+	for _, rm := range c.migrations {
+		if !c.isRedisApplied(ctx, rm.Name()) {
+			continue
+		}
+		items, err := rm.PlanCleanup(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("plan cleanup %s: %w", rm.Name(), err)
+		}
+		if items > 0 {
+			out = append(out, RedisCleanupPlan{Name: rm.Name(), Description: rm.Description(), Items: items})
+		}
+	}
+	return out, nil
+}
+
+// PlanCleanupFor returns the cleanup plan of one Redis migration. A
+// migration that is not applicable has nothing to clean up (Items 0); one
+// that is not applied yet returns ErrNotApplied.
+func (c *Coordinator) PlanCleanupFor(ctx context.Context, name string) (RedisCleanupPlan, error) {
+	rm, err := c.findRedisMigration(name)
+	if err != nil {
+		return RedisCleanupPlan{}, err
+	}
+	plan := RedisCleanupPlan{Name: rm.Name(), Description: rm.Description()}
+	status, reason := c.redisStatus(ctx, rm.Name())
+	switch status {
+	case StatusNotApplicable:
+		plan.NotApplicableReason = reason
+		return plan, nil
+	case StatusApplied:
+	default:
+		return plan, fmt.Errorf("redis/%s: %w", rm.Name(), ErrNotApplied)
+	}
+	items, err := rm.PlanCleanup(ctx)
+	if err != nil {
+		return plan, fmt.Errorf("plan cleanup %s: %w", rm.Name(), err)
+	}
+	plan.Items = items
+	return plan, nil
+}
+
+// Cleanup runs the Cleanup of one applied Redis migration under the Redis
+// migration lock.
+func (c *Coordinator) Cleanup(ctx context.Context, name string) error {
+	rm, err := c.findRedisMigration(name)
+	if err != nil {
+		return err
+	}
+	if !c.isRedisApplied(ctx, rm.Name()) {
+		return fmt.Errorf("redis/%s: %w", rm.Name(), ErrNotApplied)
+	}
+
+	lock := c.newRedisLock()
+	ok, err := lock.AttemptLock(ctx)
+	if err != nil {
+		return fmt.Errorf("acquire redis migration lock: %w", err)
+	}
+	if !ok {
+		return fmt.Errorf("acquire redis migration lock: lock already held")
+	}
+	defer func() {
+		if _, err := lock.Unlock(ctx); err != nil {
+			c.logger.Warn("failed to release redis migration lock", zap.Error(err))
+		}
+	}()
+
+	if err := rm.Cleanup(ctx, &migratorredis.State{MigrationName: rm.Name(), Phase: "applied"}); err != nil {
+		return fmt.Errorf("cleanup %s: %w", rm.Name(), err)
+	}
+	c.logger.Info("redis migration cleaned up", zap.String("migration", rm.Name()))
+	return nil
+}
+
+func (c *Coordinator) findRedisMigration(name string) (migratorredis.Migration, error) {
+	name = strings.TrimPrefix(name, "redis/")
+	for _, rm := range c.migrations {
+		if rm.Name() == name {
+			return rm, nil
+		}
+	}
+	return nil, fmt.Errorf("redis/%s: %w", name, ErrUnknownMigration)
+}
+
 // Unlock force-clears the Redis migration lock. SQL migrations use
 // PostgreSQL advisory locks which release automatically when the
 // connection closes, so there is no SQL-side unlock operation.
diff --git a/internal/migrator/coordinator/coordinator_test.go b/internal/migrator/coordinator/coordinator_test.go
index f8049ce2f..47c10df03 100644
--- a/internal/migrator/coordinator/coordinator_test.go
+++ b/internal/migrator/coordinator/coordinator_test.go
@@ -35,6 +35,9 @@ type fakeMigration struct {
 	applyCalled  bool
 	verifyCalled bool
 	verifyValid  bool
+
+	cleanupItems  int
+	cleanupCalled bool
 }
 
 func newFakeMigration(name string, version int) *fakeMigration {
@@ -92,8 +95,10 @@ func (m *fakeMigration) Verify(ctx context.Context, state *migratorredis.State)
 	}, nil
 }
 
-func (m *fakeMigration) PlanCleanup(ctx context.Context) (int, error) { return 0, nil }
+func (m *fakeMigration) PlanCleanup(ctx context.Context) (int, error) { return m.cleanupItems, nil }
 func (m *fakeMigration) Cleanup(ctx context.Context, state *migratorredis.State) error {
+	m.cleanupCalled = true
+	m.cleanupItems = 0
 	return nil
 }
 
@@ -342,3 +347,73 @@ func TestCoordinator_DeploymentID_Scoping(t *testing.T) {
 	assert.Equal(t, "applied", status,
 		"migration key should be scoped to deployment_id prefix")
 }
+
+func TestCoordinator_Cleanup(t *testing.T) {
+	ctx := context.Background()
+	setup := func(t *testing.T) (*Coordinator, *fakeMigration, *fakeMigration, *fakeMigration) {
+		applied := newFakeMigration("001_applied", 1)
+		applied.cleanupItems = 3
+		notApplicable := newFakeMigration("002_not_applicable", 2)
+		notApplicable.applicable = false
+		notApplicable.notAppReason = "not needed"
+		notApplicable.cleanupItems = 5
+		other := newFakeMigration("003_other", 3)
+		other.cleanupItems = 2
+		c, _, cleanup := newTestCoordinator(t, applied, notApplicable, other)
+		t.Cleanup(cleanup)
+		require.NoError(t, c.Apply(ctx, ApplyOptions{}))
+		return c, applied, notApplicable, other
+	}
+
+	t.Run("plan lists applied migrations with data left", func(t *testing.T) {
+		c, _, _, other := setup(t)
+		other.cleanupItems = 0
+		plans, err := c.PlanCleanup(ctx)
+		require.NoError(t, err)
+		require.Len(t, plans, 1)
+		assert.Equal(t, "001_applied", plans[0].Name)
+		assert.Equal(t, 3, plans[0].Items)
+	})
+
+	t.Run("runs only the named migration", func(t *testing.T) {
+		c, applied, notApplicable, other := setup(t)
+		plan, err := c.PlanCleanupFor(ctx, "redis/001_applied")
+		require.NoError(t, err)
+		assert.Equal(t, 3, plan.Items)
+
+		require.NoError(t, c.Cleanup(ctx, "001_applied"))
+		assert.True(t, applied.cleanupCalled)
+		assert.False(t, notApplicable.cleanupCalled)
+		assert.False(t, other.cleanupCalled)
+	})
+
+	t.Run("not applicable has nothing to clean up", func(t *testing.T) {
+		c, _, _, _ := setup(t)
+		plan, err := c.PlanCleanupFor(ctx, "002_not_applicable")
+		require.NoError(t, err)
+		assert.Equal(t, "not needed", plan.NotApplicableReason)
+		assert.Equal(t, 0, plan.Items)
+		assert.ErrorIs(t, c.Cleanup(ctx, "002_not_applicable"), ErrNotApplied)
+	})
+
+	t.Run("pending and unknown migrations", func(t *testing.T) {
+		pending := newFakeMigration("001_pending", 1)
+		c, _, cleanup := newTestCoordinator(t, pending)
+		defer cleanup()
+		_, err := c.PlanCleanupFor(ctx, "001_pending")
+		assert.ErrorIs(t, err, ErrNotApplied)
+		assert.ErrorIs(t, c.Cleanup(ctx, "001_pending"), ErrNotApplied)
+		assert.False(t, pending.cleanupCalled)
+		_, err = c.PlanCleanupFor(ctx, "999_nope")
+		assert.ErrorIs(t, err, ErrUnknownMigration)
+	})
+
+	t.Run("fails while the migration lock is held", func(t *testing.T) {
+		c, applied, _, _ := setup(t)
+		ok, err := c.newRedisLock().AttemptLock(ctx)
+		require.NoError(t, err)
+		require.True(t, ok)
+		assert.Error(t, c.Cleanup(ctx, "001_applied"))
+		assert.False(t, applied.cleanupCalled)
+	})
+}
diff --git a/internal/migrator/coordinator/types.go b/internal/migrator/coordinator/types.go
index 0e5eb4bcc..f6c57b8db 100644
--- a/internal/migrator/coordinator/types.go
+++ b/internal/migrator/coordinator/types.go
@@ -133,3 +133,14 @@ type RedisVerifyResult struct {
 	Issues       []string
 	VerifiedAt   time.Time
 }
+
+// RedisCleanupPlan describes old data an applied Redis migration can remove.
+type RedisCleanupPlan struct {
+	Name        string
+	Description string
+	// Items is the migration's own count of what Cleanup removes (e.g. keys).
+	Items int
+	// NotApplicableReason is set when the migration is not applicable for
+	// this configuration, so there is nothing to clean up.
+	NotApplicableReason string
+}
diff --git a/internal/migrator/migrations/redis.go b/internal/migrator/migrations/redis.go
index efee79f4f..e3ae94018 100644
--- a/internal/migrator/migrations/redis.go
+++ b/internal/migrator/migrations/redis.go
@@ -11,6 +11,7 @@ import (
 	migration_001 "github.com/hookdeck/outpost/internal/migrator/migratorredis/001_hash_tags"
 	migration_002 "github.com/hookdeck/outpost/internal/migrator/migratorredis/002_timestamps"
 	migration_003 "github.com/hookdeck/outpost/internal/migrator/migratorredis/003_entity"
+	migration_004 "github.com/hookdeck/outpost/internal/migrator/migratorredis/004_rsmq_hash_tags"
 )
 
 // MigrationFactory creates a migration instance with the given client, logger, and deployment ID.
@@ -29,6 +30,9 @@ var registeredMigrations = []MigrationFactory{
 	func(client redis.Client, logger migratorredis.Logger, deploymentID string) migratorredis.Migration {
 		return migration_003.New(client, logger, deploymentID)
 	},
+	func(client redis.Client, logger migratorredis.Logger, deploymentID string) migratorredis.Migration {
+		return migration_004.New(client, logger, deploymentID)
+	},
 }
 
 // AllRedisMigrations returns all registered migrations instantiated with the given client and logger.
diff --git a/internal/migrator/migratorredis/004_rsmq_hash_tags/README.md b/internal/migrator/migratorredis/004_rsmq_hash_tags/README.md
new file mode 100644
index 000000000..8bfe9b4c7
--- /dev/null
+++ b/internal/migrator/migratorredis/004_rsmq_hash_tags/README.md
@@ -0,0 +1,53 @@
+# Migration 004: RSMQ Hash Tags
+
+Moves scheduled retries from the untagged rsmq queue keys to hash-tagged keys.
+
+## Overview
+
+rsmq stores a queue in two keys: a sorted set of message IDs scored by due time, and a hash with the queue settings and message bodies. Every receive, send and delete touches both in one script or transaction. On Redis Cluster the untagged keys hash to different slots, so those operations fail with CROSSSLOT. The hash tag keeps both keys of a queue in one slot.
+
+**Key format change** (both `deliverymq-retry` and `deliverymq-retry-dlq`):
+- `rsmq:deliverymq-retry` → `rsmq:{deliverymq-retry}`
+- `rsmq:deliverymq-retry:Q` → `rsmq:{deliverymq-retry}:Q`
+
+With `DEPLOYMENT_ID` (e.g. `dp_001`), the deployment ID also goes inside the tag:
+- `dp_001:rsmq:deliverymq-retry` → `dp_001:rsmq:{dp_001:deliverymq-retry}`
+- `dp_001:rsmq:deliverymq-retry:Q` → `dp_001:rsmq:{dp_001:deliverymq-retry}:Q`
+
+`rsmq:QUEUES` (`dp_001:rsmq:QUEUES`) is unchanged.
+
+## Migration Phases
+
+### Plan
+Counts the messages under the untagged keys of each queue.
+
+### Apply
+Per queue, a Lua script moves messages in batches of 1000 (each batch atomic):
+1. Copies the queue settings (`vt`, `delay`, `maxsize`, `created`, `modified`) to the tagged hash, only where the tagged hash doesn't have them.
+2. Moves each message: sorted set member with its score, and the hash fields `<id>`, `<id>:rc`, `<id>:fr`. Messages currently being processed (hidden) keep their score, so they become due at the same time as before.
+3. If the tagged queue already holds the same message ID, keeps the tagged one and drops the old one.
+
+The untagged hash keeps its queue settings, so instances still running the previous version see a valid, empty queue and keep working without errors. Retries they schedule from then on stay under the untagged keys until Cleanup.
+
+### Verify
+Reports messages still under the untagged keys (scheduled by instances of the previous version after Apply). They are expected until Cleanup, so Verify reports the count and still passes.
+
+### Cleanup
+Run once no instance of the previous version is left (`outpost migrate cleanup 004_rsmq_hash_tags`):
+1. Moves messages scheduled under the untagged keys since Apply, the same way as Apply (newest wins: a message ID already in the tagged queue is dropped from the untagged one).
+2. Deletes the untagged keys of both queues, in the same script as the last batch.
+
+## During a Rolling Upgrade
+
+- A retry an instance of the previous version was processing during Apply, or one it scheduled during the rollout, can be delivered twice (at-least-once). Stopping the previous version before Apply avoids it.
+- Without Cleanup, retries scheduled by the previous version during the rollout never fire.
+- Rolling back to the previous version after Apply leaves the moved retries unfired (it reads only the untagged keys) until the new version runs again.
+
+## Redis Cluster
+
+Not applicable: marked `not_applicable`. On Redis Cluster the untagged queues never worked (every receive failed with CROSSSLOT), and the untagged keys can't be read atomically.
+
+## Notes
+
+- Idempotent: running Apply or Cleanup again moves only what is left.
+- Doesn't touch any other keys.
diff --git a/internal/migrator/migratorredis/004_rsmq_hash_tags/rsmq_hash_tags.go b/internal/migrator/migratorredis/004_rsmq_hash_tags/rsmq_hash_tags.go
new file mode 100644
index 000000000..152c80194
--- /dev/null
+++ b/internal/migrator/migratorredis/004_rsmq_hash_tags/rsmq_hash_tags.go
@@ -0,0 +1,243 @@
+package migration_004_rsmq_hash_tags
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/hookdeck/outpost/internal/migrator/migratorredis"
+	"github.com/hookdeck/outpost/internal/redis"
+	goredis "github.com/redis/go-redis/v9"
+)
+
+// queues are the rsmq queues used by the delivery retry scheduler.
+var queues = []string{"deliverymq-retry", "deliverymq-retry-dlq"}
+
+// moveBatchSize bounds the messages moved per script call, so a large queue
+// doesn't block Redis for long. Each call is atomic.
+const moveBatchSize = 1000
+
+// moveScript moves up to ARGV[1] messages of one rsmq queue from the
+// untagged keys to the hash-tagged keys, keeping each message's score (due
+// time, also for messages currently being processed) and its receive count
+// and first-receive time. A message whose ID already exists in the tagged
+// queue was scheduled again by a newer instance and is dropped instead.
+// Queue settings are copied to the tagged hash only where it doesn't have
+// them; the untagged hash keeps them, so instances still running the
+// previous version see a valid, empty queue.
+// With ARGV[2] == "1", the untagged keys are deleted once the zset is empty.
+//
+// KEYS[1]: untagged zset, KEYS[2]: untagged hash, KEYS[3]: tagged zset,
+// KEYS[4]: tagged hash.
+// Returns {moved, dropped, remaining}.
+var moveScript = goredis.NewScript(`
+for _, f in ipairs({"vt", "delay", "maxsize", "created", "modified"}) do
+	local v = redis.call("HGET", KEYS[2], f)
+	if v then
+		redis.call("HSETNX", KEYS[4], f, v)
+	end
+end
+local moved, dropped = 0, 0
+local batch = redis.call("ZRANGE", KEYS[1], 0, tonumber(ARGV[1]) - 1, "WITHSCORES")
+for i = 1, #batch, 2 do
+	local id, score = batch[i], batch[i + 1]
+	local body = redis.call("HGET", KEYS[2], id)
+	if body and not redis.call("ZSCORE", KEYS[3], id) then
+		redis.call("ZADD", KEYS[3], score, id)
+		redis.call("HSET", KEYS[4], id, body)
+		for _, suffix in ipairs({":rc", ":fr"}) do
+			local v = redis.call("HGET", KEYS[2], id .. suffix)
+			if v then
+				redis.call("HSET", KEYS[4], id .. suffix, v)
+			else
+				redis.call("HDEL", KEYS[4], id .. suffix)
+			end
+		end
+		moved = moved + 1
+	else
+		dropped = dropped + 1
+	end
+	redis.call("ZREM", KEYS[1], id)
+	redis.call("HDEL", KEYS[2], id, id .. ":rc", id .. ":fr")
+end
+local remaining = redis.call("ZCARD", KEYS[1])
+if remaining == 0 and ARGV[2] == "1" then
+	redis.call("DEL", KEYS[1], KEYS[2])
+end
+return {moved, dropped, remaining}
+`)
+
+// RSMQHashTagsMigration moves the delivery retry queues from the untagged rsmq
+// keys (<ns>:<queue>, <ns>:<queue>:Q) to the hash-tagged keys
+// (<ns>:{<prefix><queue>}, <ns>:{<prefix><queue>}:Q).
+type RSMQHashTagsMigration struct {
+	client    redis.Client
+	logger    migratorredis.Logger
+	ns        string // "rsmq:" or "<deployment_id>:rsmq:"
+	tagPrefix string // "" or "<deployment_id>:"
+}
+
+var _ migratorredis.Migration = (*RSMQHashTagsMigration)(nil)
+
+// New creates the migration. deploymentID is optional.
+func New(client redis.Client, logger migratorredis.Logger, deploymentID string) *RSMQHashTagsMigration {
+	m := &RSMQHashTagsMigration{client: client, logger: logger, ns: "rsmq:"}
+	if deploymentID != "" {
+		m.ns = deploymentID + ":rsmq:"
+		m.tagPrefix = deploymentID + ":"
+	}
+	return m
+}
+
+type queueKeys struct {
+	name                   string
+	oldZset, oldHash       string
+	taggedZset, taggedHash string
+}
+
+func (m *RSMQHashTagsMigration) keys(queue string) queueKeys {
+	tagged := m.ns + "{" + m.tagPrefix + queue + "}"
+	return queueKeys{
+		name:       queue,
+		oldZset:    m.ns + queue,
+		oldHash:    m.ns + queue + ":Q",
+		taggedZset: tagged,
+		taggedHash: tagged + ":Q",
+	}
+}
+
+func (m *RSMQHashTagsMigration) Name() string { return "004_rsmq_hash_tags" }
+
+func (m *RSMQHashTagsMigration) Version() int { return 4 }
+
+func (m *RSMQHashTagsMigration) Description() string {
+	return "Move scheduled retries to hash-tagged rsmq keys (rsmq:{deliverymq-retry}) for Redis Cluster support"
+}
+
+// AutoRunnable: safe to run at any time; it only moves messages and leaves
+// the old queue usable.
+func (m *RSMQHashTagsMigration) AutoRunnable() bool { return true }
+
+func (m *RSMQHashTagsMigration) IsApplicable(ctx context.Context) (bool, string) {
+	if _, ok := m.client.(*goredis.ClusterClient); ok {
+		return false, "Not needed - Redis Cluster (retry queues under the old keys never worked there)"
+	}
+	return true, ""
+}
+
+func (m *RSMQHashTagsMigration) Plan(ctx context.Context) (*migratorredis.Plan, error) {
+	scope := make(map[string]int, len(queues))
+	total := 0
+	for _, queue := range queues {
+		n, err := m.client.ZCard(ctx, m.keys(queue).oldZset).Result()
+		if err != nil {
+			return nil, fmt.Errorf("count %s: %w", queue, err)
+		}
+		scope[queue] = int(n)
+		total += int(n)
+	}
+	return &migratorredis.Plan{
+		MigrationName:  m.Name(),
+		Description:    m.Description(),
+		Version:        "v4",
+		Timestamp:      time.Now(),
+		Scope:          scope,
+		EstimatedItems: total,
+	}, nil
+}
+
+func (m *RSMQHashTagsMigration) Apply(ctx context.Context, plan *migratorredis.Plan) (*migratorredis.State, error) {
+	state := &migratorredis.State{
+		MigrationName: m.Name(),
+		Phase:         "applied",
+		StartedAt:     time.Now(),
+		Progress:      migratorredis.Progress{TotalItems: plan.EstimatedItems},
+	}
+	for _, queue := range queues {
+		moved, dropped, err := m.move(ctx, m.keys(queue), false)
+		if err != nil {
+			return nil, err
+		}
+		state.Progress.ProcessedItems += moved
+		state.Progress.SkippedItems += dropped
+	}
+	completed := time.Now()
+	state.CompletedAt = &completed
+	return state, nil
+}
+
+// move runs moveScript until the untagged zset is empty.
+func (m *RSMQHashTagsMigration) move(ctx context.Context, k queueKeys, deleteOld bool) (moved, dropped int, err error) {
+	del := "0"
+	if deleteOld {
+		del = "1"
+	}
+	for {
+		res, err := moveScript.Run(ctx, m.client,
+			[]string{k.oldZset, k.oldHash, k.taggedZset, k.taggedHash},
+			moveBatchSize, del,
+		).Int64Slice()
+		if err != nil {
+			return moved, dropped, fmt.Errorf("move %s: %w", k.name, err)
+		}
+		moved += int(res[0])
+		dropped += int(res[1])
+		if res[2] == 0 || res[0]+res[1] == 0 {
+			break
+		}
+	}
+	if moved > 0 || dropped > 0 {
+		m.logger.LogInfo(fmt.Sprintf("%s: moved %d messages, dropped %d already in the new queue", k.name, moved, dropped))
+	}
+	return moved, dropped, nil
+}
+
+// Verify reports messages still under the untagged keys: retries scheduled by
+// instances running the previous version after Apply. They are expected until
+// Cleanup moves them, so they're reported without failing verification.
+func (m *RSMQHashTagsMigration) Verify(ctx context.Context, state *migratorredis.State) (*migratorredis.VerificationResult, error) {
+	result := &migratorredis.VerificationResult{Valid: true, Details: map[string]string{}}
+	for _, queue := range queues {
+		result.ChecksRun++
+		k := m.keys(queue)
+		n, err := m.client.ZCard(ctx, k.oldZset).Result()
+		if err != nil {
+			return nil, fmt.Errorf("count %s: %w", queue, err)
+		}
+		result.Details[queue] = fmt.Sprintf("%d messages under the old keys", n)
+		if n > 0 {
+			result.Issues = append(result.Issues, fmt.Sprintf(
+				"%d retries still under the previous key layout (%s); run 'outpost migrate cleanup %s' once no instance of the previous version is running",
+				n, k.oldZset, m.Name()))
+		}
+		result.ChecksPassed++
+	}
+	return result, nil
+}
+
+// PlanCleanup returns the number of untagged keys left (zsets and hashes).
+func (m *RSMQHashTagsMigration) PlanCleanup(ctx context.Context) (int, error) {
+	total := 0
+	for _, queue := range queues {
+		k := m.keys(queue)
+		n, err := m.client.Exists(ctx, k.oldZset, k.oldHash).Result()
+		if err != nil {
+			return 0, fmt.Errorf("check %s: %w", queue, err)
+		}
+		total += int(n)
+	}
+	return total, nil
+}
+
+// Cleanup moves messages scheduled under the untagged keys since Apply
+// (newest wins on the same ID) and deletes the untagged keys. Run it once no
+// instance of the previous version is left.
+func (m *RSMQHashTagsMigration) Cleanup(ctx context.Context, state *migratorredis.State) error {
+	for _, queue := range queues {
+		if _, _, err := m.move(ctx, m.keys(queue), true); err != nil {
+			return err
+		}
+	}
+	m.logger.LogInfo("Removed the old retry queue keys.")
+	return nil
+}
diff --git a/internal/migrator/migratorredis/004_rsmq_hash_tags/rsmq_hash_tags_test.go b/internal/migrator/migratorredis/004_rsmq_hash_tags/rsmq_hash_tags_test.go
new file mode 100644
index 000000000..5753b5d8f
--- /dev/null
+++ b/internal/migrator/migratorredis/004_rsmq_hash_tags/rsmq_hash_tags_test.go
@@ -0,0 +1,316 @@
+package migration_004_rsmq_hash_tags_test
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	migration "github.com/hookdeck/outpost/internal/migrator/migratorredis/004_rsmq_hash_tags"
+
+	"github.com/hookdeck/outpost/internal/migrator/migratorredis"
+	"github.com/hookdeck/outpost/internal/redis"
+	"github.com/hookdeck/outpost/internal/rsmq"
+	"github.com/hookdeck/outpost/internal/util/testinfra"
+	"github.com/hookdeck/outpost/internal/util/testutil"
+	goredis "github.com/redis/go-redis/v9"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const queue = "deliverymq-retry"
+
+func TestRSMQHashTags_Miniredis(t *testing.T) {
+	runSuite(t, func(t *testing.T) *redis.RedisConfig { return testutil.CreateTestRedisConfig(t) })
+}
+
+func TestRSMQHashTags_Dragonfly(t *testing.T) {
+	testinfra.Start(t)
+	runSuite(t, testinfra.NewDragonflyConfig)
+}
+
+func TestRSMQHashTags_Redis(t *testing.T) {
+	testutil.SkipUnlessCompat(t)
+	testinfra.Start(t)
+	runSuite(t, testinfra.NewRedisConfig)
+}
+
+func TestRSMQHashTags_NotApplicableOnCluster(t *testing.T) {
+	client := goredis.NewClusterClient(&goredis.ClusterOptions{Addrs: []string{"127.0.0.1:1"}})
+	defer client.Close()
+	m := migration.New(client, newLogger(t), "")
+	applicable, reason := m.IsApplicable(context.Background())
+	assert.False(t, applicable)
+	assert.Contains(t, reason, "Redis Cluster")
+}
+
+func runSuite(t *testing.T, newConfig func(t *testing.T) *redis.RedisConfig) {
+	tests := []struct {
+		name         string
+		deploymentID string
+		ns           string
+		tagPrefix    string
+		oldKey       string
+		taggedKey    string
+	}{
+		{name: "without deployment id", ns: "rsmq", oldKey: "rsmq:deliverymq-retry", taggedKey: "rsmq:{deliverymq-retry}"},
+		{name: "with deployment id", deploymentID: "dp_1", ns: "dp_1:rsmq", tagPrefix: "dp_1:", oldKey: "dp_1:rsmq:deliverymq-retry", taggedKey: "dp_1:rsmq:{dp_1:deliverymq-retry}"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := newConfig(t)
+			ctx := context.Background()
+			newEnv := func(t *testing.T) *env {
+				client, err := redis.New(ctx, cfg)
+				require.NoError(t, err)
+				t.Cleanup(func() { client.Close() })
+				require.NoError(t, client.FlushDB(ctx).Err())
+				adapter := rsmq.NewRedisAdapter(client)
+				e := &env{
+					client:    client,
+					old:       rsmq.NewRedisSMQ(adapter, tt.ns, rsmq.WithUntaggedKeys()),
+					tagged:    rsmq.NewRedisSMQ(adapter, tt.ns, rsmq.WithHashTagPrefix(tt.tagPrefix)),
+					migration: migration.New(client, newLogger(t), tt.deploymentID),
+					oldKey:    tt.oldKey,
+					taggedKey: tt.taggedKey,
+				}
+				require.NoError(t, e.old.CreateQueue(queue, 30, 0, -1))
+				require.NoError(t, e.old.CreateQueue(queue+"-dlq", 30, 0, -1))
+				return e
+			}
+
+			t.Run("apply moves messages to the tagged keys", func(t *testing.T) { testApplyMoves(t, newEnv(t)) })
+			t.Run("old layout sees an empty queue after apply", func(t *testing.T) { testOldLayoutAfterApply(t, newEnv(t)) })
+			t.Run("apply keeps existing tagged queue settings", func(t *testing.T) { testApplyKeepsTaggedSettings(t, newEnv(t)) })
+			t.Run("cleanup moves leftovers, newest wins, deletes old keys", func(t *testing.T) { testCleanup(t, newEnv(t)) })
+			t.Run("apply moves more than one batch", func(t *testing.T) { testApplyBatches(t, newEnv(t)) })
+			t.Run("apply and cleanup on an empty store", func(t *testing.T) { testEmpty(t, newEnv(t)) })
+		})
+	}
+}
+
+type env struct {
+	client    redis.Client
+	old       *rsmq.RedisSMQ
+	tagged    *rsmq.RedisSMQ
+	migration *migration.RSMQHashTagsMigration
+	oldKey    string
+	taggedKey string
+}
+
+func (e *env) apply(t *testing.T) *migratorredis.State {
+	ctx := context.Background()
+	applicable, _ := e.migration.IsApplicable(ctx)
+	require.True(t, applicable)
+	plan, err := e.migration.Plan(ctx)
+	require.NoError(t, err)
+	state, err := e.migration.Apply(ctx, plan)
+	require.NoError(t, err)
+	return state
+}
+
+func testApplyMoves(t *testing.T, e *env) {
+	ctx := context.Background()
+
+	_, err := e.old.SendMessage(queue, "due", 0, rsmq.WithMessageID(id("due")))
+	require.NoError(t, err)
+	_, err = e.old.SendMessage(queue, "later", 3600, rsmq.WithMessageID(id("later")))
+	require.NoError(t, err)
+	_, err = e.old.SendMessage(queue, "inflight", 0, rsmq.WithMessageID(id("inflight")))
+	require.NoError(t, err)
+	_, err = e.old.SendMessage(queue+"-dlq", "dead", 0, rsmq.WithMessageID(id("dead")))
+	require.NoError(t, err)
+
+	// Receive "inflight" (and "due", which comes first) with a long visibility
+	// timeout, as an instance processing it would; put "due" back.
+	for i := 0; i < 2; i++ {
+		msg, err := e.old.ReceiveMessage(queue, 3600)
+		require.NoError(t, err)
+		require.NotNil(t, msg)
+	}
+	require.NoError(t, e.old.ChangeMessageVisibility(queue, id("due"), 0))
+
+	scores := map[string]float64{}
+	for _, name := range []string{"due", "later", "inflight"} {
+		s, err := e.client.ZScore(ctx, e.oldKey, id(name)).Result()
+		require.NoError(t, err)
+		scores[id(name)] = s
+	}
+	oldRC, err := e.client.HGet(ctx, e.oldKey+":Q", id("inflight")+":rc").Result()
+	require.NoError(t, err)
+	oldFR, err := e.client.HGet(ctx, e.oldKey+":Q", id("inflight")+":fr").Result()
+	require.NoError(t, err)
+
+	plan, err := e.migration.Plan(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, 4, plan.EstimatedItems)
+	assert.Equal(t, 3, plan.Scope[queue])
+	assert.Equal(t, 1, plan.Scope[queue+"-dlq"])
+
+	state, err := e.migration.Apply(ctx, plan)
+	require.NoError(t, err)
+	assert.Equal(t, 4, state.Progress.ProcessedItems)
+
+	for msgID, score := range scores {
+		s, err := e.client.ZScore(ctx, e.taggedKey, msgID).Result()
+		require.NoError(t, err, msgID)
+		assert.Equal(t, score, s, msgID)
+	}
+	rc, err := e.client.HGet(ctx, e.taggedKey+":Q", id("inflight")+":rc").Result()
+	require.NoError(t, err)
+	assert.Equal(t, oldRC, rc)
+	fr, err := e.client.HGet(ctx, e.taggedKey+":Q", id("inflight")+":fr").Result()
+	require.NoError(t, err)
+	assert.Equal(t, oldFR, fr)
+
+	attrs, err := e.tagged.GetQueueAttributes(queue)
+	require.NoError(t, err)
+	assert.Equal(t, uint(30), attrs.Vt)
+	assert.Equal(t, uint64(3), attrs.Msgs)
+	assert.Equal(t, uint64(2), attrs.HiddenMsgs)
+
+	msg, err := e.tagged.ReceiveMessage(queue, 30)
+	require.NoError(t, err)
+	require.NotNil(t, msg)
+	assert.Equal(t, id("due"), msg.ID)
+	assert.Equal(t, "due", msg.Message)
+
+	dead, err := e.tagged.ReceiveMessage(queue+"-dlq", 30)
+	require.NoError(t, err)
+	require.NotNil(t, dead)
+	assert.Equal(t, "dead", dead.Message)
+
+	n, err := e.client.ZCard(ctx, e.oldKey).Result()
+	require.NoError(t, err)
+	assert.Equal(t, int64(0), n)
+
+	result, err := e.migration.Verify(ctx, state)
+	require.NoError(t, err)
+	assert.True(t, result.Valid, result.Issues)
+}
+
+func testOldLayoutAfterApply(t *testing.T, e *env) {
+	_, err := e.old.SendMessage(queue, "msg", 0, rsmq.WithMessageID(id("a")))
+	require.NoError(t, err)
+	e.apply(t)
+
+	msg, err := e.old.ReceiveMessage(queue, 30)
+	require.NoError(t, err)
+	assert.Nil(t, msg)
+	poll, err := e.old.ReceiveMessagePoll(queue, 30)
+	require.NoError(t, err)
+	assert.Nil(t, poll.Message)
+	assert.False(t, poll.HasNext)
+	// Acking a message that was moved finds nothing.
+	assert.ErrorIs(t, e.old.DeleteMessage(queue, id("a")), rsmq.ErrMessageNotFound)
+
+	// It can still schedule; cleanup picks those up later.
+	_, err = e.old.SendMessage(queue, "msg", 0, rsmq.WithMessageID(id("b")))
+	require.NoError(t, err)
+	msg, err = e.old.ReceiveMessage(queue, 30)
+	require.NoError(t, err)
+	require.NotNil(t, msg)
+	assert.Equal(t, id("b"), msg.ID)
+}
+
+func testApplyKeepsTaggedSettings(t *testing.T, e *env) {
+	require.NoError(t, e.tagged.CreateQueue(queue, 99, 0, -1))
+	e.apply(t)
+	attrs, err := e.tagged.GetQueueAttributes(queue)
+	require.NoError(t, err)
+	assert.Equal(t, uint(99), attrs.Vt)
+	// The dlq had no tagged queue: created with the old settings.
+	attrs, err = e.tagged.GetQueueAttributes(queue + "-dlq")
+	require.NoError(t, err)
+	assert.Equal(t, uint(30), attrs.Vt)
+}
+
+func testCleanup(t *testing.T, e *env) {
+	ctx := context.Background()
+	state := e.apply(t)
+
+	// During the rollout, an instance of the previous version schedules two
+	// retries; a new instance schedules one of them again.
+	_, err := e.old.SendMessage(queue, "old-a", 0, rsmq.WithMessageID(id("a")))
+	require.NoError(t, err)
+	_, err = e.old.SendMessage(queue, "old-b", 0, rsmq.WithMessageID(id("b")))
+	require.NoError(t, err)
+	_, err = e.tagged.SendMessage(queue, "new-a", 3600, rsmq.WithMessageID(id("a")))
+	require.NoError(t, err)
+
+	// Leftovers are expected until cleanup: reported, not a failure.
+	result, err := e.migration.Verify(ctx, state)
+	require.NoError(t, err)
+	assert.True(t, result.Valid)
+	assert.Equal(t, 2, result.ChecksPassed)
+	require.Len(t, result.Issues, 1)
+	assert.Contains(t, result.Issues[0], "2 retries still under the previous key layout")
+	assert.Contains(t, result.Issues[0], "once no instance of the previous version is running")
+
+	n, err := e.migration.PlanCleanup(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, 3, n) // retry zset + hash, dlq hash (its zset is empty)
+
+	require.NoError(t, e.migration.Cleanup(ctx, state))
+
+	body, err := e.client.HGet(ctx, e.taggedKey+":Q", id("a")).Result()
+	require.NoError(t, err)
+	assert.Equal(t, "new-a", body)
+	body, err = e.client.HGet(ctx, e.taggedKey+":Q", id("b")).Result()
+	require.NoError(t, err)
+	assert.Equal(t, "old-b", body)
+
+	exists, err := e.client.Exists(ctx, e.oldKey, e.oldKey+":Q", e.oldKey+"-dlq", e.oldKey+"-dlq:Q").Result()
+	require.NoError(t, err)
+	assert.Equal(t, int64(0), exists)
+
+	n, err = e.migration.PlanCleanup(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, 0, n)
+	result, err = e.migration.Verify(ctx, state)
+	require.NoError(t, err)
+	assert.True(t, result.Valid, result.Issues)
+	assert.Empty(t, result.Issues)
+
+	msg, err := e.tagged.ReceiveMessage(queue, 30)
+	require.NoError(t, err)
+	require.NotNil(t, msg)
+	assert.Equal(t, id("b"), msg.ID)
+}
+
+func testApplyBatches(t *testing.T, e *env) {
+	ctx := context.Background()
+	const count = 2500
+	for i := 0; i < count; i++ {
+		_, err := e.old.SendMessage(queue, "msg", 0, rsmq.WithMessageID(id(fmt.Sprintf("m%05d", i))))
+		require.NoError(t, err)
+	}
+	state := e.apply(t)
+	assert.Equal(t, count, state.Progress.ProcessedItems)
+	n, err := e.client.ZCard(ctx, e.taggedKey).Result()
+	require.NoError(t, err)
+	assert.Equal(t, int64(count), n)
+	n, err = e.client.ZCard(ctx, e.oldKey).Result()
+	require.NoError(t, err)
+	assert.Equal(t, int64(0), n)
+}
+
+func testEmpty(t *testing.T, e *env) {
+	ctx := context.Background()
+	require.NoError(t, e.client.FlushDB(ctx).Err())
+	state := e.apply(t)
+	assert.Equal(t, 0, state.Progress.ProcessedItems)
+	require.NoError(t, e.migration.Cleanup(ctx, state))
+	keys, err := e.client.Keys(ctx, "*").Result()
+	require.NoError(t, err)
+	assert.Empty(t, keys)
+}
+
+// id pads name to a valid rsmq message ID.
+func id(name string) string {
+	return name + strings.Repeat("0", 32-len(name))
+}
+
+func newLogger(t *testing.T) migratorredis.Logger {
+	return migratorredis.NewLoggerAdapter(testutil.CreateTestLogger(t), false)
+}
diff --git a/internal/rsmq/rsmq.go b/internal/rsmq/rsmq.go
index 1b65a0f53..59468dd67 100644
--- a/internal/rsmq/rsmq.go
+++ b/internal/rsmq/rsmq.go
@@ -106,9 +106,11 @@ type RedisClient interface {
 
 // RedisSMQ is the client of rsmq to execute queue and message operations
 type RedisSMQ struct {
-	client RedisClient
-	ns     string
-	logger *logging.Logger
+	client    RedisClient
+	ns        string
+	logger    *logging.Logger
+	tagPrefix string
+	untagged  bool
 }
 
 // QueueAttributes contains some attributes and stats of queue
@@ -167,8 +169,36 @@ type Client interface {
 
 var _ Client = (*RedisSMQ)(nil)
 
+// Option configures a RedisSMQ.
+type Option func(*RedisSMQ)
+
+// WithLogger sets the logger. A nil logger disables logging.
+func WithLogger(logger *logging.Logger) Option {
+	return func(rsmq *RedisSMQ) {
+		rsmq.logger = logger
+	}
+}
+
+// WithHashTagPrefix prepends prefix to the queue name inside the hash tag, so
+// queues with the same name in different namespaces (e.g. one per deployment
+// on a shared Redis Cluster) map to different slots.
+func WithHashTagPrefix(prefix string) Option {
+	return func(rsmq *RedisSMQ) {
+		rsmq.tagPrefix = prefix
+	}
+}
+
+// WithUntaggedKeys uses the key layout from before queue names were
+// hash-tagged (<ns>:<qname>, <ns>:<qname>:Q), as earlier versions did; tests
+// use it to play those versions. Its keys span two Redis Cluster slots.
+func WithUntaggedKeys() Option {
+	return func(rsmq *RedisSMQ) {
+		rsmq.untagged = true
+	}
+}
+
 // NewRedisSMQ creates and returns new rsmq client
-func NewRedisSMQ(client RedisClient, ns string, logger ...*logging.Logger) *RedisSMQ {
+func NewRedisSMQ(client RedisClient, ns string, opts ...Option) *RedisSMQ {
 	if client == nil {
 		panic("")
 	}
@@ -180,15 +210,12 @@ func NewRedisSMQ(client RedisClient, ns string, logger ...*logging.Logger) *Redi
 		ns += ":"
 	}
 
-	var l *logging.Logger
-	if len(logger) > 0 {
-		l = logger[0]
-	}
-
 	rsmq := &RedisSMQ{
 		client: client,
 		ns:     ns,
-		logger: l,
+	}
+	for _, opt := range opts {
+		opt(rsmq)
 	}
 
 	client.ScriptLoad(scriptPopMessage)
@@ -198,6 +225,16 @@ func NewRedisSMQ(client RedisClient, ns string, logger ...*logging.Logger) *Redi
 	return rsmq
 }
 
+// queueKey returns the key of the queue's zset; its hash is queueKey + ":Q".
+// The queue name is hash-tagged so both keys map to the same Redis Cluster
+// slot, as the scripts and transactions touching both require.
+func (rsmq *RedisSMQ) queueKey(qname string) string {
+	if rsmq.untagged {
+		return rsmq.ns + qname
+	}
+	return rsmq.ns + "{" + rsmq.tagPrefix + qname + "}"
+}
+
 // CreateQueue creates a new queue with given attributes
 // to create new queue with default attributes:
 //
@@ -236,7 +273,7 @@ func (rsmq *RedisSMQ) CreateQueue(qname string, vt uint, delay uint, maxsize int
 		return err
 	}
 
-	key := rsmq.ns + qname + q
+	key := rsmq.queueKey(qname) + q
 
 	tx := rsmq.client.TxPipeline()
 	r := tx.HSetNX(key, "vt", vt)
@@ -310,7 +347,7 @@ func (rsmq *RedisSMQ) CreateQueue(qname string, vt uint, delay uint, maxsize int
 }
 
 func (rsmq *RedisSMQ) getQueue(qname string, uid bool) (*queueDef, error) {
-	key := rsmq.ns + qname + q
+	key := rsmq.queueKey(qname) + q
 
 	tx := rsmq.client.TxPipeline()
 
@@ -362,7 +399,7 @@ func (rsmq *RedisSMQ) GetQueueAttributes(qname string) (*QueueAttributes, error)
 		return nil, err
 	}
 
-	key := rsmq.ns + qname
+	key := rsmq.queueKey(qname)
 
 	tx := rsmq.client.TxPipeline()
 	hmGetSliceCmd := tx.HMGet(key+q, "vt", "delay", "maxsize", "totalrecv", "totalsent", "created", "modified")
@@ -433,7 +470,7 @@ func (rsmq *RedisSMQ) SetQueueAttributes(qname string, vt uint, delay uint, maxs
 		return nil, err
 	}
 
-	key := rsmq.ns + qname + q
+	key := rsmq.queueKey(qname) + q
 
 	tx := rsmq.client.TxPipeline()
 	tx.HSet(key, "modified", queue.ts)
@@ -458,7 +495,7 @@ func (rsmq *RedisSMQ) DeleteQueue(qname string) error {
 		return err
 	}
 
-	key := rsmq.ns + qname
+	key := rsmq.queueKey(qname)
 
 	tx := rsmq.client.TxPipeline()
 	r := tx.Del(key + q)
@@ -507,7 +544,7 @@ func (rsmq *RedisSMQ) SendMessage(qname string, message string, delay uint, opts
 		return "", ErrMessageTooLong
 	}
 
-	key := rsmq.ns + qname
+	key := rsmq.queueKey(qname)
 
 	// Use custom ID if provided, otherwise use generated one
 	messageID := queue.uid
@@ -560,7 +597,7 @@ func (rsmq *RedisSMQ) ReceiveMessagePoll(qname string, vt uint) (PollResult, err
 		vtArg = strconv.FormatUint(uint64(vt), 10)
 	}
 
-	key := rsmq.ns + qname
+	key := rsmq.queueKey(qname)
 	hashKey := key + q // key + ":Q"
 
 	evalCmd := rsmq.client.EvalSha(hashReceiveMessage, []string{key, hashKey}, vtArg)
@@ -624,7 +661,7 @@ func (rsmq *RedisSMQ) PopMessage(qname string) (*QueueMessage, error) {
 		return nil, err
 	}
 
-	key := rsmq.ns + qname
+	key := rsmq.queueKey(qname)
 	hashKey := key + q // key + ":Q"
 
 	t := strconv.FormatUint(queue.ts, 10)
@@ -711,7 +748,7 @@ func (rsmq *RedisSMQ) ChangeMessageVisibility(qname string, id string, vt uint)
 		return err
 	}
 
-	key := rsmq.ns + qname
+	key := rsmq.queueKey(qname)
 	t := strconv.FormatUint(queue.ts+uint64(vt)*1000, 10)
 
 	evalCmd := rsmq.client.EvalSha(hashChangeMessageVisibility, []string{key}, id, t)
@@ -738,7 +775,7 @@ func (rsmq *RedisSMQ) DeleteMessage(qname string, id string) error {
 		return err
 	}
 
-	key := rsmq.ns + qname
+	key := rsmq.queueKey(qname)
 
 	tx := rsmq.client.TxPipeline()
 	zremIntCmd := tx.ZRem(key, id)
diff --git a/internal/rsmq/rsmq_test.go b/internal/rsmq/rsmq_test.go
index d5f9e9526..7bfe11194 100644
--- a/internal/rsmq/rsmq_test.go
+++ b/internal/rsmq/rsmq_test.go
@@ -898,3 +898,53 @@ func (s *RSMQSuite) TestSendMessageWithCustomID() {
 		}
 	})
 }
+
+func (s *RSMQSuite) TestQueueKeysHashTagged() {
+	t := s.T()
+	qname := "que"
+
+	assert.NoError(t, s.rsmq.CreateQueue(qname, UnsetVt, UnsetDelay, UnsetMaxsize))
+	_, err := s.rsmq.SendMessage(qname, "msg", 0)
+	assert.NoError(t, err)
+
+	n, err := s.client.Exists("test:{que}", "test:{que}:Q").Result()
+	assert.NoError(t, err)
+	assert.Equal(t, int64(2), n)
+	n, err = s.client.Exists("test:que", "test:que:Q").Result()
+	assert.NoError(t, err)
+	assert.Equal(t, int64(0), n)
+
+	t.Run("untagged client uses the legacy layout", func(t *testing.T) {
+		legacy := NewRedisSMQ(s.client, "test", WithUntaggedKeys())
+		assert.NoError(t, legacy.CreateQueue(qname, UnsetVt, UnsetDelay, UnsetMaxsize))
+		_, err := legacy.SendMessage(qname, "legacy", 0)
+		assert.NoError(t, err)
+
+		n, err := s.client.Exists("test:que", "test:que:Q").Result()
+		assert.NoError(t, err)
+		assert.Equal(t, int64(2), n)
+
+		msg, err := legacy.ReceiveMessage(qname, UnsetVt)
+		assert.NoError(t, err)
+		if assert.NotNil(t, msg) {
+			assert.Equal(t, "legacy", msg.Message)
+		}
+	})
+
+	t.Run("hash tag prefix goes inside the tag", func(t *testing.T) {
+		prefixed := NewRedisSMQ(s.client, "dp_1:test", WithHashTagPrefix("dp_1:"))
+		assert.NoError(t, prefixed.CreateQueue(qname, UnsetVt, UnsetDelay, UnsetMaxsize))
+		_, err := prefixed.SendMessage(qname, "prefixed", 0)
+		assert.NoError(t, err)
+
+		n, err := s.client.Exists("dp_1:test:{dp_1:que}", "dp_1:test:{dp_1:que}:Q").Result()
+		assert.NoError(t, err)
+		assert.Equal(t, int64(2), n)
+
+		msg, err := prefixed.ReceiveMessage(qname, UnsetVt)
+		assert.NoError(t, err)
+		if assert.NotNil(t, msg) {
+			assert.Equal(t, "prefixed", msg.Message)
+		}
+	})
+}