[3/4 messagequeue] Wire MQ_TENANTS through services and tests

## Summary

### Why?

The MySQL subscriber only discovers partitions for configured tenants. Service processes and compose stacks must pass the same list they already use as queue names, or Subscribe fails at startup and tests talk to an unsharded process.

### What?

- Parse MQ_TENANTS in gateway, orchestrator, and runway wiring and reject configs whose queue names do not match.
- Set MQ_TENANTS in service compose files, e2e harnesses, and SubmitQueue integration suites.

## Test Plan

✅ `go test` TestValidateConfiguredQueueTenants, TestValidateProfileQueueTenants, TestValidateMergeQueueTenants, and service/messagequeue

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

# Conflicts:
#	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 95ba176
# Last command done (1 command done):
#    pick cc23648 # [3/4 messagequeue] Wire MQ_TENANTS through services and tests
# No commands remaining.
# You are currently rebasing branch 'preetam/mq-tenant-wiring' on '95ba1764'.
#
# Changes to be committed:
#	modified:   service/runway/server/BUILD.bazel
#	modified:   service/runway/server/config_test.go
#	modified:   service/runway/server/docker-compose.yml
#	modified:   service/runway/server/main.go
#	modified:   service/stovepipe/docker-compose.yml
#	modified:   service/submitqueue/docker-compose.yml
#	modified:   service/submitqueue/gateway/server/BUILD.bazel
#	modified:   service/submitqueue/gateway/server/docker-compose.yml
#	modified:   service/submitqueue/gateway/server/main.go
#	modified:   service/submitqueue/gateway/server/main_test.go
#	modified:   service/submitqueue/gateway/server/queues.yaml
#	modified:   service/submitqueue/orchestrator/server/BUILD.bazel
#	modified:   service/submitqueue/orchestrator/server/config_test.go
#	modified:   service/submitqueue/orchestrator/server/docker-compose.yml
#	modified:   service/submitqueue/orchestrator/server/main.go
#	modified:   test/e2e/runway/harness_test.go
#	modified:   test/e2e/runway/suite_test.go
#	modified:   test/e2e/stovepipe/harness_test.go
#	modified:   test/e2e/stovepipe/suite_test.go
#	modified:   test/integration/submitqueue/core/consumer/consumer_test.go
#	modified:   test/integration/submitqueue/gateway/BUILD.bazel
#	modified:   test/integration/submitqueue/gateway/suite_test.go
#

Author: behinddwalls

service/runway/server/BUILD.bazel

@@ -38,6 +38,7 @@ go_library(
         "//runway/extension/merger/fake:go_default_library",
         "//runway/extension/merger/git:go_default_library",
         "//runway/extension/merger/noop:go_default_library",
+        "//service/messagequeue:go_default_library",
         "@com_github_go_sql_driver_mysql//:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
         "@in_gopkg_yaml_v3//:go_default_library",

service/runway/server/config_test.go

@@ -28,6 +28,12 @@ import (
 	mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb"
 )
 
+func TestValidateMergeQueueTenants(t *testing.T) {
+	cfg := mergeConfig{Queues: []namedQueueMergeConfig{{Name: "queue-a"}}}
+	require.NoError(t, validateMergeQueueTenants([]string{"queue-a", "queue-b"}, cfg))
+	require.Error(t, validateMergeQueueTenants([]string{"queue-b"}, cfg))
+}
+
 // writeConfig writes a merge config file and returns its path.
 func writeConfig(t *testing.T, contents string) string {
 	t.Helper()

service/runway/server/docker-compose.yml

@@ -53,6 +53,7 @@ services:
       # Level for the queue's own logs; info by default so its per-message
       # chatter does not bury the rest of the service at debug.
       - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-}
+      - MQ_TENANTS=${MQ_TENANTS:-test-queue,e2e-runway/merge,e2e-runway/check,e2e-runway/failed,e2e-runway/dlq,e2e-runway/undecodable}
       - HOSTNAME=runway-dev
     depends_on:
       mysql-queue:

service/runway/server/main.go

@@ -52,6 +52,7 @@ import (
 	"github.com/uber/submitqueue/runway/extension/merger/fake"
 	gitmerger "github.com/uber/submitqueue/runway/extension/merger/git"
 	"github.com/uber/submitqueue/runway/extension/merger/noop"
+	servicemq "github.com/uber/submitqueue/service/messagequeue"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/reflection"
@@ -140,11 +141,17 @@ func run() error {
 	}
 	defer queueDB.Close()
 
+	tenants, err := servicemq.ParseRequiredTenants(os.Getenv("MQ_TENANTS"))
+	if err != nil {
+		return fmt.Errorf("failed to configure queue subscribers: %w", err)
+	}
+
 	mysqlQueue, err := queueMySQL.NewQueue(queueMySQL.Params{
 		DB:           queueDB,
 		Logger:       logger,
 		LogLevel:     os.Getenv("QUEUE_LOG_LEVEL"),
 		MetricsScope: scope.SubScope("queue"),
+		Tenants:      tenants,
 	})
 	if err != nil {
 		return fmt.Errorf("failed to create queue: %w", err)
@@ -170,7 +177,7 @@ func run() error {
 
 	primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry, newPrimaryErrorProcessor(), gate)
 
-	mergerFactory, err := newMergerFactory(ctx, logger, scope.SubScope("merger"))
+	mergerFactory, err := newMergerFactory(ctx, logger, scope.SubScope("merger"), tenants)
 	if err != nil {
 		return fmt.Errorf("failed to create merger factory: %w", err)
 	}
@@ -327,7 +334,7 @@ func newPrimaryErrorProcessor() errs.ErrorProcessor {
 // The fake is reachable only through MERGER, never through the configuration
 // file: an implementation whose outcomes are steered by markers in a change URI
 // has no business being selectable by a production config.
-func newMergerFactory(ctx context.Context, logger *zap.Logger, scope tally.Scope) (merger.Factory, error) {
+func newMergerFactory(ctx context.Context, logger *zap.Logger, scope tally.Scope, tenants []string) (merger.Factory, error) {
 	startup, err := resolveMergerStartupConfig(logger)
 	if err != nil {
 		return nil, err
@@ -339,12 +346,15 @@ func newMergerFactory(ctx context.Context, logger *zap.Logger, scope tally.Scope
 		// demand without a git checkout. Never production.
 		logger.Info("MERGER=fake; using marker-driven fake merger for every queue")
 		return &fakeMergerFactory{seq: new(atomic.Uint64)}, nil
-	case "noop":
+	case mergerTypeNoop:
 		logger.Info("MERGER=noop; using noop merger for every queue")
 		return &noopMergerFactory{seq: new(atomic.Uint64)}, nil
 	}
 
 	cfg := startup.targets
+	if err := validateMergeQueueTenants(tenants, cfg); err != nil {
+		return nil, fmt.Errorf("failed to validate queue tenants: %w", err)
+	}
 
 	// The git runtime is resolved only when something actually needs it, so a
 	// deployment running nothing but the noop merger does not require git to be
@@ -420,6 +430,14 @@ func resolveMergerStartupConfig(logger *zap.Logger) (mergerStartupConfig, error)
 	return mergerStartupConfig{selection: selection, targets: cfg}, nil
 }
 
+func validateMergeQueueTenants(tenants []string, cfg mergeConfig) error {
+	mergeQueueNames := make([]string, 0, len(cfg.Queues))
+	for _, queue := range cfg.Queues {
+		mergeQueueNames = append(mergeQueueNames, queue.Name)
+	}
+	return servicemq.ValidateTenantSubset("MQ_TENANTS", tenants, "merge config", mergeQueueNames)
+}
+
 // loadMergeConfigFromEnv reads the merge configuration file when one is
 // configured, and otherwise reconstructs the equivalent single-queue
 // configuration from the MERGE_* environment.

service/stovepipe/docker-compose.yml

@@ -70,6 +70,7 @@ services:
       # Level for the queue's own logs; info by default so its per-message
       # chatter does not bury the rest of the service at debug.
       - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-}
+      - MQ_TENANTS=${MQ_TENANTS:-monorepo/main,monorepo/release,monorepo/slow?buildrunner-fake=build-slow}
       - HOSTNAME=stovepipe-dev
     depends_on:
       mysql-app:

service/submitqueue/docker-compose.yml

@@ -76,6 +76,7 @@ services:
       # Level for the queue's own logs; info by default so its per-message
       # chatter does not bury the rest of the service at debug.
       - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-}
+      - MQ_TENANTS=${MQ_TENANTS:-test-queue,e2e-test-queue,e2e-cancel-queue,e2e-chain-queue,e2e-redelivery-queue,e2e-strand-queue,e2e-conflict-error-queue,e2e-git-queue,demo-queue,e2e-respeculate-queue,file-overlap-queue}
       # Path to YAML queue configuration baked into the image
       - QUEUE_CONFIG_PATH=/app/queues.yaml
       # Stable subscriber name for the request-log consumer
@@ -107,6 +108,7 @@ services:
       # Level for the queue's own logs; info by default so its per-message
       # chatter does not bury the rest of the service at debug.
       - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-}
+      - MQ_TENANTS=${MQ_TENANTS:-test-queue,e2e-test-queue,e2e-cancel-queue,e2e-chain-queue,e2e-redelivery-queue,e2e-strand-queue,e2e-conflict-error-queue,e2e-git-queue,demo-queue,e2e-respeculate-queue,file-overlap-queue}
       - HOSTNAME=orchestrator-dev
       # Consumer-gate state shared with the host (see header comment)
       - CONSUMER_GATE_DIR=/var/submitqueue/consumergate
@@ -138,6 +140,7 @@ services:
       # Level for the queue's own logs; info by default so its per-message
       # chatter does not bury the rest of the service at debug.
       - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-}
+      - MQ_TENANTS=${MQ_TENANTS:-test-queue,e2e-test-queue,e2e-cancel-queue,e2e-chain-queue,e2e-redelivery-queue,e2e-strand-queue,e2e-conflict-error-queue,e2e-git-queue,demo-queue,e2e-respeculate-queue,file-overlap-queue}
       - HOSTNAME=runway-dev
       # Consumer-gate state shared with the host (see header comment)
       - CONSUMER_GATE_DIR=/var/submitqueue/consumergate

service/submitqueue/gateway/server/BUILD.bazel

@@ -23,6 +23,7 @@ go_library(
         "//platform/extension/counter/mysql:go_default_library",
         "//platform/extension/messagequeue:go_default_library",
         "//platform/extension/messagequeue/mysql:go_default_library",
+        "//service/messagequeue:go_default_library",
         "//service/submitqueue/gateway/server/mapper:go_default_library",
         "//submitqueue/core/request:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
@@ -77,6 +78,7 @@ go_test(
     deps = [
         "//submitqueue/gateway/controller:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
         "@org_golang_google_grpc//codes:go_default_library",
         "@org_golang_google_grpc//status:go_default_library",
     ],

service/submitqueue/gateway/server/docker-compose.yml

@@ -65,6 +65,7 @@ services:
       # Level for the queue's own logs; info by default so its per-message
       # chatter does not bury the rest of the service at debug.
       - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-}
+      - MQ_TENANTS=${MQ_TENANTS:-test-queue,e2e-test-queue,e2e-cancel-queue,e2e-chain-queue,e2e-redelivery-queue,e2e-strand-queue,e2e-conflict-error-queue,e2e-git-queue,demo-queue,e2e-respeculate-queue,file-overlap-queue}
       # Path to YAML queue configuration baked into the image
       - QUEUE_CONFIG_PATH=/app/queues.yaml
       # Stable subscriber name for the request-log consumer

service/submitqueue/gateway/server/main.go

@@ -40,6 +40,7 @@ import (
 	mysqlcounter "github.com/uber/submitqueue/platform/extension/counter/mysql"
 	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
 	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
+	servicemq "github.com/uber/submitqueue/service/messagequeue"
 	"github.com/uber/submitqueue/service/submitqueue/gateway/server/mapper"
 	requestcore "github.com/uber/submitqueue/submitqueue/core/request"
 	"github.com/uber/submitqueue/submitqueue/core/topickey"
@@ -241,12 +242,38 @@ func run() error {
 	}
 	defer queueDB.Close()
 
-	// Initialize queue
+	// Load queue configurations from YAML. Path is required so the gateway
+	// can reject requests for unknown queues at the edge.
+	queueConfigPath := os.Getenv("QUEUE_CONFIG_PATH")
+	if queueConfigPath == "" {
+		return fmt.Errorf("QUEUE_CONFIG_PATH environment variable is required")
+	}
+	queueConfigs, err := yamlqueueconfig.NewStore(queueConfigPath)
+	if err != nil {
+		return fmt.Errorf("failed to load queue configs: %w", err)
+	}
+	configuredQueues, err := queueConfigs.List(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to list queue configs: %w", err)
+	}
+	configuredQueueNames := make([]string, 0, len(configuredQueues))
+	for _, q := range configuredQueues {
+		configuredQueueNames = append(configuredQueueNames, q.Name)
+	}
+	tenants, err := servicemq.ParseRequiredTenants(os.Getenv("MQ_TENANTS"))
+	if err != nil {
+		return fmt.Errorf("failed to configure queue subscribers: %w", err)
+	}
+	if err := validateConfiguredQueueTenants(tenants, configuredQueueNames); err != nil {
+		return fmt.Errorf("failed to validate queue tenants: %w", err)
+	}
+
 	mysqlQueue, err := queueMySQL.NewQueue(queueMySQL.Params{
 		DB:           queueDB,
 		Logger:       logger,
 		LogLevel:     os.Getenv("QUEUE_LOG_LEVEL"),
 		MetricsScope: scope.SubScope("queue"),
+		Tenants:      tenants,
 	})
 	if err != nil {
 		return fmt.Errorf("failed to create queue: %w", err)
@@ -314,16 +341,6 @@ func run() error {
 	if err != nil {
 		return fmt.Errorf("failed to create storage: %w", err)
 	}
-	// Load queue configurations from YAML. Path is required so the gateway
-	// can reject requests for unknown queues at the edge.
-	queueConfigPath := os.Getenv("QUEUE_CONFIG_PATH")
-	if queueConfigPath == "" {
-		return fmt.Errorf("QUEUE_CONFIG_PATH environment variable is required")
-	}
-	queueConfigs, err := yamlqueueconfig.NewStore(queueConfigPath)
-	if err != nil {
-		return fmt.Errorf("failed to load queue configs: %w", err)
-	}
 
 	// Create controllers and wrap them for gRPC. Every store is queue-scoped and
 	// resolves through the factory adapter; land/cancel/log share one materializer.
@@ -441,6 +458,10 @@ func run() error {
 	return err
 }
 
+func validateConfiguredQueueTenants(tenants, configuredQueueNames []string) error {
+	return servicemq.ValidateTenantSetsEqual("MQ_TENANTS", tenants, "QUEUE_CONFIG_PATH", configuredQueueNames)
+}
+
 // newConsumerGate enables the file-backed consumer gate only when
 // CONSUMER_GATE_DIR is explicitly configured. The file implementation is for
 // E2E and single-host development; normal service deployments use the no-op

service/submitqueue/gateway/server/main_test.go

@@ -19,11 +19,23 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"github.com/uber/submitqueue/submitqueue/gateway/controller"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
+func TestValidateConfiguredQueueTenants(t *testing.T) {
+	require.NoError(t, validateConfiguredQueueTenants(
+		[]string{"queue-a", "queue-b"},
+		[]string{"queue-b", "queue-a"},
+	))
+	require.Error(t, validateConfiguredQueueTenants(
+		[]string{"queue-a"},
+		[]string{"queue-a", "queue-b"},
+	))
+}
+
 func TestGatewayStatusError(t *testing.T) {
 	tests := []struct {
 		name string