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

service/runway/server/BUILD.bazel

@@ -37,6 +37,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

@@ -26,6 +26,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

@@ -52,6 +52,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

@@ -51,6 +51,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"
@@ -135,11 +136,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)
@@ -171,7 +178,7 @@ func run() error {
 		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)
 	}
@@ -315,7 +322,7 @@ func run() error {
 // 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) {
 	switch impl := strings.ToLower(strings.TrimSpace(os.Getenv("MERGER"))); impl {
 	case "fake":
 		// Marker-driven outcomes, for e2e tests that need Runway to fail on
@@ -335,6 +342,9 @@ func newMergerFactory(ctx context.Context, logger *zap.Logger, scope tally.Scope
 	if err != nil {
 		return nil, err
 	}
+	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
@@ -385,6 +395,14 @@ func newMergerFactory(ctx context.Context, logger *zap.Logger, scope tally.Scope
 	return mergerRegistry{byQueue: byQueue, fallback: fallback}, 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

service/submitqueue/gateway/server/queues.yaml

@@ -32,3 +32,6 @@ queues:
   # second request lands as a batch depending on the first. e2e uses that to
   # exercise speculation across an unresolved dependency.
   - name: e2e-respeculate-queue
+  # Exercises path-overlap conflict analysis in the built-in orchestrator
+  # profile.
+  - name: file-overlap-queue

service/submitqueue/orchestrator/server/BUILD.bazel

@@ -33,6 +33,7 @@ go_library(
         "//platform/githubactions:go_default_library",
         "//platform/http:go_default_library",
         "//platform/pipeline:go_default_library",
+        "//service/messagequeue:go_default_library",
         "//submitqueue/core/changeset:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/buildrunner:go_default_library",

service/submitqueue/orchestrator/server/config_test.go

@@ -37,6 +37,12 @@ import (
 	"github.com/uber/submitqueue/submitqueue/extension/speculation/speculator"
 )
 
+func TestValidateProfileQueueTenants(t *testing.T) {
+	cfg := profilesConfig{Queues: []namedQueueProfileConfig{{Name: "queue-a"}}}
+	require.NoError(t, validateProfileQueueTenants([]string{"queue-a", "queue-b"}, cfg))
+	require.Error(t, validateProfileQueueTenants([]string{"queue-b"}, cfg))
+}
+
 func writeProfiles(t *testing.T, contents string) string {
 	t.Helper()
 	path := filepath.Join(t.TempDir(), "profiles.yaml")

service/submitqueue/orchestrator/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-conflict-error-queue,file-overlap-queue}
       - HOSTNAME=orchestrator-dev
     depends_on:
       mysql-app:

service/submitqueue/orchestrator/server/main.go

@@ -42,6 +42,7 @@ import (
 	hooknoop "github.com/uber/submitqueue/platform/extension/hook/noop"
 	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
 	"github.com/uber/submitqueue/platform/pipeline"
+	servicemq "github.com/uber/submitqueue/service/messagequeue"
 	"github.com/uber/submitqueue/submitqueue/core/changeset"
 	"github.com/uber/submitqueue/submitqueue/extension/storage"
 	mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql"
@@ -156,12 +157,28 @@ func run() error {
 	}
 	defer queueDB.Close()
 
-	// Initialize queue
+	// Build per-queue extension profiles (host-private). Each queue resolves
+	// to its own set of extension implementations (conflict analyzer, …),
+	// falling back to a baseline profile for queues without an explicit entry.
+	storageFty := storageFactory{backend: store}
+	profilesCfg, err := loadProfilesConfigFromEnv(logger)
+	if err != nil {
+		return fmt.Errorf("failed to load extension profiles: %w", err)
+	}
+	tenants, err := servicemq.ParseRequiredTenants(os.Getenv("MQ_TENANTS"))
+	if err != nil {
+		return fmt.Errorf("failed to configure queue subscribers: %w", err)
+	}
+	if err := validateProfileQueueTenants(tenants, profilesCfg); 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)
@@ -176,14 +193,6 @@ func run() error {
 		subscriberName = fmt.Sprintf("orchestrator-%d", time.Now().Unix())
 	}
 
-	// Build per-queue extension profiles (host-private). Each queue resolves
-	// to its own set of extension implementations (conflict analyzer, …),
-	// falling back to a baseline profile for queues without an explicit entry.
-	storageFty := storageFactory{backend: store}
-	profilesCfg, err := loadProfilesConfigFromEnv(logger)
-	if err != nil {
-		return fmt.Errorf("failed to load extension profiles: %w", err)
-	}
 	profiles, err := newProfiles(ctx, logger, scope, changeset.New(storageFty), storageFty, profilesCfg)
 	if err != nil {
 		return fmt.Errorf("failed to build profiles: %w", err)
@@ -307,6 +316,14 @@ func run() error {
 	return err
 }
 
+func validateProfileQueueTenants(tenants []string, cfg profilesConfig) error {
+	profileQueueNames := make([]string, 0, len(cfg.Queues))
+	for _, queue := range cfg.Queues {
+		profileQueueNames = append(profileQueueNames, queue.Name)
+	}
+	return servicemq.ValidateTenantSubset("MQ_TENANTS", tenants, "extension profiles", profileQueueNames)
+}
+
 // 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

test/e2e/runway/harness_test.go

@@ -167,7 +167,10 @@ func (s *RunwayE2ESuite) publish(topic string, request *runwaymq.MergeRequest) {
 func (s *RunwayE2ESuite) publishRaw(topic, id, partitionKey string, payload []byte) {
 	t := s.T()
 
-	msg := entityqueue.NewMessage(id, payload, partitionKey, nil)
+	msg := entityqueue.NewMessage(id, payload, partitionKey, map[string]string{
+		entityqueue.MetadataKeyQueueName: partitionKey,
+	})
+	msg.Tenant = partitionKey
 	require.NoError(t, s.queue.Publisher().Publish(s.ctx, topic, msg),
 		"failed to publish %s to %s", id, topic)
 	s.log.Logf("published %s to %s (partition %s)", id, topic, partitionKey)