refactor(messagequeue): make publish tenant explicit

## Summary

### Why?

Tenant selects the physical MQ shard and is therefore a first-class publish input, not metadata to infer implicitly. Parsing MQ_TENANTS is service wiring behavior rather than a responsibility of the MySQL backend.

### What?

- Require tenant explicitly in platform publish and hook APIs, propagate it to queue_name metadata, and reject conflicting metadata.
- Pass each producer's authoritative queue through SubmitQueue, Runway, Stovepipe, hooks, and tests independently from partition keys.
- Move required-tenant parsing into a shared service/messagequeue wiring package.

## Test Plan

✅ Focused unit tests across publish, hook, service wiring, Runway, Stovepipe, and SubmitQueue controllers
✅ `./tool/bazel build //...`
✅ `make fmt && make gazelle`

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: behinddwalls

doc/rfc/messagequeue-tenant-sharding.md

@@ -76,7 +76,7 @@ Fair-share, orphan sweep, and idle-lease release run per `(tenant, topic)`, not
 
 ## Publish
 
-`platform/publish` stamps `Message.Tenant` from context metadata (`queue_name`). Empty tenant on publish is rejected. `PartitionKey` is unchanged.
+Every `platform/publish` call supplies tenant explicitly. The package stamps `Message.Tenant` and mirrors it into `queue_name` delivery metadata, rejecting empty tenants and conflicting caller metadata. `PartitionKey` is unchanged.
 
 ## Wiring
 

platform/extension/messagequeue/mysql/BUILD.bazel

@@ -16,7 +16,6 @@ go_library(
         "stores.go",
         "subscriber.go",
         "subscriber_heartbeat_store.go",
-        "tenants.go",
     ],
     importpath = "github.com/uber/submitqueue/platform/extension/messagequeue/mysql",
     visibility = ["//visibility:public"],
@@ -43,7 +42,6 @@ go_test(
         "sql_test.go",
         "subscriber_heartbeat_store_test.go",
         "subscriber_test.go",
-        "tenants_test.go",
     ],
     embed = [":go_default_library"],
     deps = [

platform/extension/messagequeue/mysql/tenants.go

@@ -23,9 +23,9 @@ import (
 	"github.com/uber/submitqueue/platform/publish"
 )
 
-// Publish sends one hook event to the domain's hook topic, partitioned by
-// partitionKey. The topic key is not a parameter: a domain runs a single hook
-// topic, and the caller's registry is what binds that key to a wire topic.
+// Publish sends one hook event to the domain's hook topic for tenant,
+// partitioned by partitionKey. The topic key is not a parameter: a domain runs
+// a single hook topic, and the caller's registry binds that key to a wire topic.
 //
 // The event id is the message id, so a redelivery republishing the same event
 // dedups into the original message instead of enqueuing a second one. Callers
@@ -37,6 +37,7 @@ import (
 func Publish(
 	ctx context.Context,
 	registry consumer.TopicRegistry,
+	tenant string,
 	event *basehook.HookEvent,
 	partitionKey string,
 ) error {
@@ -50,7 +51,7 @@ func Publish(
 	}
 
 	if err := publish.Message(
-		ctx, registry, basehook.TopicKeyHook, publish.IntentID(event.GetId()), body, partitionKey,
+		ctx, registry, basehook.TopicKeyHook, tenant, publish.IntentID(event.GetId()), body, partitionKey,
 	); err != nil {
 		return fmt.Errorf("failed to publish hook event %s: %w", event.GetId(), err)
 	}

platform/hook/publisher.go

@@ -31,6 +31,7 @@ import (
 const (
 	testEventID      = "stovepipe/validation.repository.recorded/request/7/0"
 	testPartitionKey = "request/7"
+	testTenant       = "monorepo/main"
 )
 
 func testEvent() *basehook.HookEvent {
@@ -69,12 +70,14 @@ func TestPublish(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	registry, published := registryWithHookTopic(t, ctrl, nil)
 
-	require.NoError(t, Publish(context.Background(), registry, testEvent(), testPartitionKey))
+	require.NoError(t, Publish(context.Background(), registry, testTenant, testEvent(), testPartitionKey))
 
 	// The event id is the message id, so a redelivery republishing the same
 	// event dedups into the original message.
 	assert.Equal(t, testEventID, published.ID)
 	assert.Equal(t, testPartitionKey, published.PartitionKey)
+	assert.Equal(t, testTenant, published.Tenant)
+	assert.Equal(t, testTenant, published.Metadata[entityqueue.MetadataKeyQueueName])
 
 	decoded := &basehook.HookEvent{}
 	require.NoError(t, basehook.Unmarshal(published.Payload, decoded))
@@ -105,7 +108,7 @@ func TestPublish_RejectsMalformedEvent(t *testing.T) {
 			require.NoError(t, err)
 
 			// No Publish expectation: a malformed event must not reach the queue.
-			require.Error(t, Publish(context.Background(), registry, tt.event, testPartitionKey))
+			require.Error(t, Publish(context.Background(), registry, testTenant, tt.event, testPartitionKey))
 		})
 	}
 }
@@ -114,12 +117,12 @@ func TestPublish_PropagatesPublishFailure(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	registry, _ := registryWithHookTopic(t, ctrl, errors.New("boom"))
 
-	require.Error(t, Publish(context.Background(), registry, testEvent(), testPartitionKey))
+	require.Error(t, Publish(context.Background(), registry, testTenant, testEvent(), testPartitionKey))
 }
 
 func TestPublish_FailsWhenHookTopicIsUnregistered(t *testing.T) {
 	registry, err := consumer.NewTopicRegistry(nil)
 	require.NoError(t, err)
 
-	require.Error(t, Publish(context.Background(), registry, testEvent(), testPartitionKey))
+	require.Error(t, Publish(context.Background(), registry, testTenant, testEvent(), testPartitionKey))
 }

platform/hook/publisher_test.go

@@ -35,11 +35,11 @@ import (
 	"github.com/uber/submitqueue/platform/consumer"
 )
 
-// Message publishes payload to the topic registered for key. Allowlisted
-// delivery context is propagated as message metadata.
+// Message publishes payload to the topic registered for key. Tenant selects
+// the shard and is propagated as the delivery's queue-name metadata.
 //
 // msgID selects the dedup behavior, so the caller must choose it deliberately.
-// The queue deduplicates on (topic, partition key, message ID) against every
+// The queue deduplicates on (tenant, topic, partition key, message ID) against every
 // row it has not garbage-collected yet, consumed ones included — a window with
 // no upper bound on a busy partition. A publish that collides is reported as a
 // success and writes nothing, and nothing retries it.
@@ -48,16 +48,18 @@ import (
 // this particular message exists for. A retry of the same cause then dedups,
 // which is what makes redelivery safe, while a new cause about the same entity
 // can never be swallowed by an older row.
-func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string) error {
-	return MessageWithMetadata(ctx, registry, key, msgID, payload, partitionKey, nil)
+func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, tenant, msgID string, payload []byte, partitionKey string) error {
+	return MessageWithMetadata(ctx, registry, key, tenant, msgID, payload, partitionKey, nil)
 }
 
 // MessageWithMetadata is Message with side-band message metadata (headers/attributes)
 // attached to the delivery. Use it to carry diagnostic context that is not part of
 // the payload — the backend persists and redelivers metadata alongside the message.
-// Allowlisted delivery context, currently only the queue name, is propagated unless
-// the caller supplies that metadata key explicitly.
-func MessageWithMetadata(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string, metadata map[string]string) error {
+// Tenant is also propagated as the delivery's queue-name context.
+func MessageWithMetadata(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, tenant, msgID string, payload []byte, partitionKey string, metadata map[string]string) error {
+	if tenant == "" {
+		return fmt.Errorf("tenant is required")
+	}
 	q, ok := registry.Queue(key)
 	if !ok {
 		return fmt.Errorf("no queue registered for topic key %s", key)
@@ -67,29 +69,18 @@ func MessageWithMetadata(ctx context.Context, registry consumer.TopicRegistry, k
 		return fmt.Errorf("no topic name registered for topic key %s", key)
 	}
 
-	msg := entityqueue.NewMessage(msgID, payload, partitionKey, metadataFromContext(ctx, metadata))
-	if tenant := msg.Metadata[entityqueue.MetadataKeyQueueName]; tenant != "" {
-		msg.Tenant = tenant
-	} else if queueName, ok := entityqueue.QueueName(ctx); ok {
-		msg.Tenant = queueName
+	if queueName, exists := metadata[entityqueue.MetadataKeyQueueName]; exists && queueName != tenant {
+		return fmt.Errorf("queue-name metadata %q does not match tenant %q", queueName, tenant)
 	}
-	return q.Publisher().Publish(ctx, topicName, msg)
-}
-
-func metadataFromContext(ctx context.Context, metadata map[string]string) map[string]string {
 	metadata = maps.Clone(metadata)
-	if _, exists := metadata[entityqueue.MetadataKeyQueueName]; exists {
-		return metadata
-	}
-	queueName, ok := entityqueue.QueueName(ctx)
-	if !ok || queueName == "" {
-		return metadata
-	}
 	if metadata == nil {
 		metadata = make(map[string]string)
 	}
-	metadata[entityqueue.MetadataKeyQueueName] = queueName
-	return metadata
+	metadata[entityqueue.MetadataKeyQueueName] = tenant
+
+	msg := entityqueue.NewMessage(msgID, payload, partitionKey, metadata)
+	msg.Tenant = tenant
+	return q.Publisher().Publish(ctx, topicName, msg)
 }
 
 // IntentID names the occasion to publish rather than the entity published

platform/publish/publish.go

@@ -55,15 +55,16 @@ func TestMessage(t *testing.T) {
 			return nil
 		})
 
-	err := Message(context.Background(), registry, testKey, "msg-1", []byte("payload"), "partition-1")
+	err := Message(context.Background(), registry, testKey, "tenant-1", "msg-1", []byte("payload"), "partition-1")
 	require.NoError(t, err)
+	assert.Equal(t, "tenant-1", published.Tenant)
 	assert.Equal(t, "msg-1", published.ID)
 	assert.Equal(t, []byte("payload"), published.Payload)
 	assert.Equal(t, "partition-1", published.PartitionKey)
-	assert.Empty(t, published.Metadata)
+	assert.Equal(t, "tenant-1", published.Metadata[entityqueue.MetadataKeyQueueName])
 }
 
-func TestMessage_PropagatesQueueNameFromContext(t *testing.T) {
+func TestMessage_PropagatesTenantAsQueueName(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	registry, publisher := newTestRegistry(t, ctrl)
 
@@ -75,8 +76,7 @@ func TestMessage_PropagatesQueueNameFromContext(t *testing.T) {
 			return nil
 		})
 
-	ctx := entityqueue.WithQueueName(context.Background(), "monorepo/main")
-	require.NoError(t, Message(ctx, registry, testKey, "msg-1", []byte("payload"), "partition-1"))
+	require.NoError(t, Message(context.Background(), registry, testKey, "monorepo/main", "msg-1", []byte("payload"), "partition-1"))
 	assert.Equal(t, "monorepo/main", published.Metadata[entityqueue.MetadataKeyQueueName])
 	assert.Equal(t, "monorepo/main", published.Tenant)
 }
@@ -94,8 +94,7 @@ func TestMessageWithMetadata_MergesContextWithoutMutatingInput(t *testing.T) {
 		})
 
 	metadata := map[string]string{"failure_reason": "build failed"}
-	ctx := entityqueue.WithQueueName(context.Background(), "monorepo/main")
-	require.NoError(t, MessageWithMetadata(ctx, registry, testKey, "msg-1", []byte("payload"), "partition-1", metadata))
+	require.NoError(t, MessageWithMetadata(context.Background(), registry, testKey, "monorepo/main", "msg-1", []byte("payload"), "partition-1", metadata))
 	assert.Equal(t, map[string]string{
 		"failure_reason":                 "build failed",
 		entityqueue.MetadataKeyQueueName: "monorepo/main",
@@ -104,30 +103,28 @@ func TestMessageWithMetadata_MergesContextWithoutMutatingInput(t *testing.T) {
 	assert.Equal(t, map[string]string{"failure_reason": "build failed"}, metadata)
 }
 
-func TestMessageWithMetadata_ExplicitQueueNameWins(t *testing.T) {
+func TestMessageWithMetadata_RejectsQueueNameDifferentFromTenant(t *testing.T) {
 	ctrl := gomock.NewController(t)
-	registry, publisher := newTestRegistry(t, ctrl)
-
-	var published entityqueue.Message
-	publisher.EXPECT().
-		Publish(gomock.Any(), "test-topic", gomock.Any()).
-		DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error {
-			published = msg
-			return nil
-		})
+	registry, _ := newTestRegistry(t, ctrl)
 
-	ctx := entityqueue.WithQueueName(context.Background(), "inbound")
 	metadata := map[string]string{entityqueue.MetadataKeyQueueName: "outbound"}
-	require.NoError(t, MessageWithMetadata(ctx, registry, testKey, "msg-1", []byte("payload"), "partition-1", metadata))
-	assert.Equal(t, "outbound", published.Metadata[entityqueue.MetadataKeyQueueName])
-	assert.Equal(t, "outbound", published.Tenant)
+	err := MessageWithMetadata(context.Background(), registry, testKey, "inbound", "msg-1", []byte("payload"), "partition-1", metadata)
+	require.Error(t, err)
 }
 
 func TestMessage_UnregisteredKey(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	registry, _ := newTestRegistry(t, ctrl)
 
-	err := Message(context.Background(), registry, "unregistered-key", "msg-1", []byte("payload"), "partition-1")
+	err := Message(context.Background(), registry, "unregistered-key", "tenant-1", "msg-1", []byte("payload"), "partition-1")
+	require.Error(t, err)
+}
+
+func TestMessage_RequiresTenant(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	registry, _ := newTestRegistry(t, ctrl)
+
+	err := Message(context.Background(), registry, testKey, "", "msg-1", []byte("payload"), "partition-1")
 	require.Error(t, err)
 }
 

platform/publish/publish_test.go

@@ -165,7 +165,7 @@ func (c *Controller) publish(ctx context.Context, result *runwaymq.MergeResult,
 	}
 
 	if err := publish.Message(ctx, c.registry, c.signalTopicKey,
-		publish.IntentID(result.GetId(), "dlq"), payload, partitionKey); err != nil {
+		result.GetQueueName(), publish.IntentID(result.GetId(), "dlq"), payload, partitionKey); err != nil {
 		return fmt.Errorf("failed to publish message: %w", err)
 	}
 

runway/controller/dlq/dlq.go

@@ -112,6 +112,7 @@ func TestProcess_DecodableRepublishesFailure(t *testing.T) {
 	require.Len(t, *published, 1)
 	got := (*published)[0]
 	assert.Equal(t, "merge-signal", got.topic)
+	assert.Equal(t, testQueue, got.msg.Tenant)
 
 	result := &runwaymq.MergeResult{}
 	require.NoError(t, runwaymq.Unmarshal(got.msg.Payload, result))

runway/controller/dlq/dlq_test.go

@@ -153,7 +153,7 @@ func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, result
 		return fmt.Errorf("failed to serialize merge result: %w", err)
 	}
 
-	if err := publish.Message(ctx, c.registry, key, publish.IntentID(result.GetId()), payload, partitionKey); err != nil {
+	if err := publish.Message(ctx, c.registry, key, result.GetQueueName(), publish.IntentID(result.GetId()), payload, partitionKey); err != nil {
 		return fmt.Errorf("failed to publish message: %w", err)
 	}