fix(messagequeue): harden tenant shard isolation

## Summary

### Why?

The initial tenant-sharding implementation could exceed InnoDB's composite-key limit, infer incomplete tenant lists from extension overrides, and let one tenant's storage failure block unrelated tenants. Stovepipe could also accept work for a tenant its subscribers would never discover.

### What?

- Use readable ASCII VARCHAR identifiers for tenants, topics, and consumer names while preserving utf8mb4 partition keys and message IDs.
- Validate backend identifier contracts before database access and require an authoritative MQ_TENANTS list for consumer services.
- Reject unconfigured Stovepipe queues before side effects.
- Isolate discovery and shutdown failures per tenant and aggregate errors after all tenants are attempted.
- Narrow the queueshard AUTO_INCREMENT exception and add schema, cross-tenant ACK/GC, and DLQ isolation coverage.

## Test Plan

✅ `./tool/bazel test //platform/extension/messagequeue/mysql/... //tool/linter/queueshard/... --test_output=errors`
✅ `./tool/bazel test //test/integration/extension/messagequeue/mysql:go_default_test --test_output=errors`
✅ `./tool/bazel build //service/runway/server/... //service/stovepipe/server/... //service/submitqueue/orchestrator/server/...`
✅ `make fmt && make gazelle`

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

# Conflicts:
#	service/runway/server/config.go
#	service/runway/server/main.go

# Please enter the commit message for your changes. Lines starting
# with '#' will be kept; you may remove them yourself if you want to.
# An empty message aborts the commit.
#
# interactive rebase in progress; onto c9ee7a8
# Last commands done (2 commands done):
#    pick 8940d26 # feat(messagequeue): shard MySQL queue tables by tenant
#    pick c79649f # fix(messagequeue): harden tenant shard isolation
# No commands remaining.
# You are currently rebasing branch 'preetam/mq-tenant-shard' on 'c9ee7a84'.
#
# Changes to be committed:
#	modified:   doc/rfc/messagequeue-tenant-sharding.md
#	modified:   platform/extension/messagequeue/mysql/BUILD.bazel
#	new file:   platform/extension/messagequeue/mysql/identifier.go
#	modified:   platform/extension/messagequeue/mysql/publisher.go
#	modified:   platform/extension/messagequeue/mysql/publisher_test.go
#	modified:   platform/extension/messagequeue/mysql/schema/queue_delivery_state.sql
#	modified:   platform/extension/messagequeue/mysql/schema/queue_messages.sql
#	modified:   platform/extension/messagequeue/mysql/schema/queue_offsets.sql
#	modified:   platform/extension/messagequeue/mysql/schema/queue_partition_leases.sql
#	modified:   platform/extension/messagequeue/mysql/schema/queue_subscriber_heartbeats.sql
#	modified:   platform/extension/messagequeue/mysql/subscriber.go
#	modified:   platform/extension/messagequeue/mysql/subscriber_test.go
#	modified:   platform/extension/messagequeue/mysql/tenants.go
#	new file:   platform/extension/messagequeue/mysql/tenants_test.go
#	modified:   service/runway/server/config.go
#	modified:   service/runway/server/docker-compose.yml
#	modified:   service/runway/server/main.go
#	modified:   service/stovepipe/server/main.go
#	modified:   service/submitqueue/docker-compose.yml
#	modified:   service/submitqueue/orchestrator/server/docker-compose.yml
#	modified:   service/submitqueue/orchestrator/server/main.go
#	modified:   stovepipe/controller/ingest.go
#	modified:   stovepipe/controller/ingest_test.go
#	modified:   test/integration/extension/messagequeue/mysql/BUILD.bazel
#	modified:   test/integration/extension/messagequeue/mysql/queue_test.go
#	new file:   test/integration/extension/messagequeue/mysql/tenant_isolation_test.go
#	modified:   tool/linter/queueshard/main.go
#	modified:   tool/linter/queueshard/main_test.go
#

Author: behinddwalls

doc/rfc/messagequeue-tenant-sharding.md

@@ -31,13 +31,14 @@ The MQ schema does not use `queue` — that word is overloaded (SubmitQueue doma
 
 ## Schema
 
-Every table's primary key leads with `tenant`. Secondary indexes that do not lead with `tenant` are removed.
+Every table's primary key leads with `tenant`. Tenant, topic, consumer-group, and subscriber identifiers use `VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin`; partition keys and message IDs use explicit `VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`. This keeps operational identifiers readable, preserves unrestricted UTF-8 ordering keys and IDs, and keeps the largest composite key within InnoDB's 3072-byte limit. The backend validates each contract before database access. Secondary indexes that do not lead with `tenant` are removed except `queue_messages.idx_offset`, which InnoDB requires because the `AUTO_INCREMENT offset` column must be leftmost in an index.
 
 ### `queue_messages`
 
 - PK: `(tenant, topic, partition_key, offset)`
 - Unique: `(tenant, topic, partition_key, id)`
-- `offset` is per-partition, not global; fetch is `WHERE tenant=? AND topic=? AND partition_key=? AND offset>? ORDER BY offset`
+- Required InnoDB index: `idx_offset (offset)` for the `AUTO_INCREMENT` column
+- `offset` is allocated from a shard-wide monotonic sequence and used as an ordering cursor within each partition; fetch is `WHERE tenant=? AND topic=? AND partition_key=? AND offset>? ORDER BY offset`
 
 ### `queue_delivery_state`
 
@@ -63,23 +64,23 @@ DLQ moves rewrite `topic` to `original + suffix` and keep `tenant` + `partition_
 
 Today partition discovery runs `SELECT DISTINCT partition_key FROM queue_messages WHERE topic=?`, which scatter-gathers across all Vitess shards.
 
-The subscriber takes a configured tenant list (SubmitQueue: queue names from YAML). Discovery becomes:
+The subscriber takes an explicit configured tenant list from `MQ_TENANTS`. Consumer processes reject an empty list at startup; Stovepipe also rejects ingest requests for names outside the list. Discovery becomes:
 
 ```sql
 SELECT DISTINCT partition_key FROM queue_messages
 WHERE tenant = ? AND topic = ?
 ORDER BY partition_key
 ```
 
-Fair-share, orphan sweep, and idle-lease release run per `(tenant, topic)`, not across all tenants on a topic.
+Fair-share, orphan sweep, and idle-lease release run per `(tenant, topic)`, not across all tenants on a topic. Discovery and shutdown attempt every configured tenant and aggregate errors so one unavailable shard does not block unrelated tenants.
 
 ## Publish
 
 `platform/publish` stamps `Message.Tenant` from context metadata (`queue_name`). Empty tenant on publish is rejected. `PartitionKey` is unchanged.
 
 ## Wiring
 
-One `extqueue.Queue` and one VTGate DSN per service. `NewQueue` / subscriber `Params` carry `Tenants []string`. Service `main.go` fills that from configured queue names.
+One `extqueue.Queue` and one VTGate DSN per service. `NewQueue` / subscriber `Params` carry `Tenants []string`. Consumer service wiring parses the authoritative comma-separated `MQ_TENANTS` list once and passes it to the backend and any ingress validation.
 
 ## Out of scope
 

platform/extension/messagequeue/mysql/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "constants.go",
         "delivery_state_store.go",
         "errors.go",
+        "identifier.go",
         "message_store.go",
         "mock_stores.go",
         "offset_store.go",
@@ -42,6 +43,7 @@ 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/identifier.go

@@ -0,0 +1,44 @@
+// Copyright (c) 2026 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mysql
+
+import (
+	"fmt"
+	"unicode/utf8"
+)
+
+const maxIdentifierLength = 255
+
+func validateASCIIIdentifier(name, value string) error {
+	if len(value) > maxIdentifierLength {
+		return fmt.Errorf("%s exceeds %d bytes", name, maxIdentifierLength)
+	}
+	for i := range len(value) {
+		if value[i] > 0x7f {
+			return fmt.Errorf("%s must contain only ASCII characters", name)
+		}
+	}
+	return nil
+}
+
+func validateTextIdentifier(name, value string) error {
+	if !utf8.ValidString(value) {
+		return fmt.Errorf("%s is not valid UTF-8", name)
+	}
+	if utf8.RuneCountInString(value) > maxIdentifierLength {
+		return fmt.Errorf("%s exceeds %d characters", name, maxIdentifierLength)
+	}
+	return nil
+}

platform/extension/messagequeue/mysql/publisher.go

@@ -60,6 +60,28 @@ func (p *publisher) Publish(ctx context.Context, topic string, message entityque
 	if message.Tenant == "" {
 		return fmt.Errorf("publish: message tenant is required")
 	}
+	for _, identifier := range []struct {
+		name  string
+		value string
+	}{
+		{name: "tenant", value: message.Tenant},
+		{name: "topic", value: topic},
+	} {
+		if err := validateASCIIIdentifier(identifier.name, identifier.value); err != nil {
+			return fmt.Errorf("publish: %w", err)
+		}
+	}
+	for _, identifier := range []struct {
+		name  string
+		value string
+	}{
+		{name: "message ID", value: message.ID},
+		{name: "partition key", value: message.PartitionKey},
+	} {
+		if err := validateTextIdentifier(identifier.name, identifier.value); err != nil {
+			return fmt.Errorf("publish: %w", err)
+		}
+	}
 
 	if err := p.messageStore.Insert(ctx, message.Tenant, topic, []entityqueue.Message{message}); err != nil {
 		return fmt.Errorf("publish message store insert error: %w", err)

platform/extension/messagequeue/mysql/publisher_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -43,6 +44,8 @@ func setupPublisherTest(t *testing.T, mockStore *MockmessageStore) extqueue.Publ
 }
 
 func TestPublisher_Publish(t *testing.T) {
+	overlong := strings.Repeat("x", maxIdentifierLength+1)
+	noStoreCall := func(*MockmessageStore) {}
 	tests := []struct {
 		name      string
 		topic     string
@@ -112,6 +115,56 @@ func TestPublisher_Publish(t *testing.T) {
 				m.EXPECT().Insert(gomock.Any(), testTenant, "topic-with-dash", gomock.Any()).Return(nil).Times(1)
 			},
 		},
+		{
+			name:      "rejects overlong tenant",
+			topic:     "test_topic",
+			messages:  []entityqueue.Message{{Tenant: overlong, ID: "msg1", PartitionKey: "part1"}},
+			wantErr:   true,
+			setupMock: noStoreCall,
+		},
+		{
+			name:      "rejects overlong topic",
+			topic:     overlong,
+			messages:  []entityqueue.Message{{Tenant: testTenant, ID: "msg1", PartitionKey: "part1"}},
+			wantErr:   true,
+			setupMock: noStoreCall,
+		},
+		{
+			name:      "rejects overlong message ID",
+			topic:     "test_topic",
+			messages:  []entityqueue.Message{{Tenant: testTenant, ID: overlong, PartitionKey: "part1"}},
+			wantErr:   true,
+			setupMock: noStoreCall,
+		},
+		{
+			name:      "rejects overlong partition key",
+			topic:     "test_topic",
+			messages:  []entityqueue.Message{{Tenant: testTenant, ID: "msg1", PartitionKey: overlong}},
+			wantErr:   true,
+			setupMock: noStoreCall,
+		},
+		{
+			name:     "accepts UTF-8 message ID and partition key",
+			topic:    "test_topic",
+			messages: []entityqueue.Message{{Tenant: testTenant, ID: "msg-é", PartitionKey: strings.Repeat("é", maxIdentifierLength)}},
+			setupMock: func(m *MockmessageStore) {
+				m.EXPECT().Insert(gomock.Any(), testTenant, "test_topic", gomock.Any()).Return(nil)
+			},
+		},
+		{
+			name:      "rejects non-ASCII tenant",
+			topic:     "test_topic",
+			messages:  []entityqueue.Message{{Tenant: "tenant-é", ID: "msg1", PartitionKey: "part1"}},
+			wantErr:   true,
+			setupMock: noStoreCall,
+		},
+		{
+			name:      "rejects non-ASCII topic",
+			topic:     "topic-é",
+			messages:  []entityqueue.Message{{Tenant: testTenant, ID: "msg1", PartitionKey: "part1"}},
+			wantErr:   true,
+			setupMock: noStoreCall,
+		},
 	}
 
 	for _, tt := range tests {

platform/extension/messagequeue/mysql/schema/queue_delivery_state.sql

@@ -9,16 +9,16 @@
 
 CREATE TABLE IF NOT EXISTS queue_delivery_state (
     -- tenant is the shard isolation identity
-    tenant VARCHAR(255) NOT NULL,
+    tenant VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- Consumer group this delivery state belongs to
-    consumer_group VARCHAR(255) NOT NULL,
+    consumer_group VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- Topic of the message
-    topic VARCHAR(255) NOT NULL,
+    topic VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- Partition key of the message
-    partition_key VARCHAR(255) NOT NULL,
+    partition_key VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
 
     -- Offset of the message in the immutable log
     message_offset BIGINT UNSIGNED NOT NULL,

platform/extension/messagequeue/mysql/schema/queue_messages.sql

@@ -5,19 +5,19 @@
 
 CREATE TABLE IF NOT EXISTS queue_messages (
     -- tenant is the shard isolation identity (SubmitQueue maps queueName here at wiring)
-    tenant VARCHAR(255) NOT NULL,
+    tenant VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- Topic identifies the pipeline stage
-    topic VARCHAR(255) NOT NULL,
+    topic VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- Partition key for distributing work across workers within a tenant
-    partition_key VARCHAR(255) NOT NULL,
+    partition_key VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
 
     -- Auto-incrementing offset for ordering within (tenant, topic, partition_key)
     offset BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
 
     -- Message identification
-    id VARCHAR(255) NOT NULL,
+    id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
 
     -- Message data
     payload BLOB NOT NULL,
@@ -31,7 +31,7 @@ CREATE TABLE IF NOT EXISTS queue_messages (
     failed_at BIGINT UNSIGNED NOT NULL,
     failure_count INT UNSIGNED NOT NULL,
     last_error TEXT NOT NULL,
-    original_topic VARCHAR(255) NOT NULL,
+    original_topic VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
     failure_detail JSON,
 
     PRIMARY KEY (tenant, topic, partition_key, offset),

platform/extension/messagequeue/mysql/schema/queue_offsets.sql

@@ -4,16 +4,16 @@
 
 CREATE TABLE IF NOT EXISTS queue_offsets (
     -- tenant is the shard isolation identity
-    tenant VARCHAR(255) NOT NULL,
+    tenant VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- Consumer group consuming the topic
-    consumer_group VARCHAR(255) NOT NULL,
+    consumer_group VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- Topic being consumed
-    topic VARCHAR(255) NOT NULL,
+    topic VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- Partition being consumed
-    partition_key VARCHAR(255) NOT NULL,
+    partition_key VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
 
     -- Last offset that was successfully acked for this partition
     offset_acked BIGINT UNSIGNED NOT NULL,

platform/extension/messagequeue/mysql/schema/queue_partition_leases.sql

@@ -4,19 +4,19 @@
 
 CREATE TABLE IF NOT EXISTS queue_partition_leases (
     -- tenant is the shard isolation identity
-    tenant VARCHAR(255) NOT NULL,
+    tenant VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- Consumer group (e.g., "orchestrator")
-    consumer_group VARCHAR(255) NOT NULL,
+    consumer_group VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- Topic being consumed
-    topic VARCHAR(255) NOT NULL,
+    topic VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- Partition that is leased
-    partition_key VARCHAR(255) NOT NULL,
+    partition_key VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
 
     -- Worker that owns the lease (e.g., "worker-1")
-    leased_by VARCHAR(255) NOT NULL,
+    leased_by VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- When lease was acquired (epoch milliseconds)
     leased_at BIGINT UNSIGNED NOT NULL,

platform/extension/messagequeue/mysql/schema/queue_subscriber_heartbeats.sql

@@ -4,16 +4,16 @@
 
 CREATE TABLE IF NOT EXISTS queue_subscriber_heartbeats (
     -- tenant is the shard isolation identity
-    tenant VARCHAR(255) NOT NULL,
+    tenant VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- consumer_group identifies the consumer group this subscriber belongs to
-    consumer_group VARCHAR(255) NOT NULL,
+    consumer_group VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- topic is the topic this subscriber is consuming from
-    topic VARCHAR(255) NOT NULL,
+    topic VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- subscriber_name uniquely identifies this subscriber within the consumer group
-    subscriber_name VARCHAR(255) NOT NULL,
+    subscriber_name VARCHAR(255) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
 
     -- heartbeat_at is the Unix timestamp in milliseconds of the last heartbeat
     heartbeat_at BIGINT UNSIGNED NOT NULL,