feat(storage)!: resolve queue-scoped storage through a factory, split off global read-model stores

api/runway/messagequeue/proto/merge.proto

@@ -126,4 +126,7 @@ message MergeResult {
     // merge populates each step's outputs with the revisions it produced; a
     // dry-run check leaves them empty.
     repeated StepResult steps = 4;
+    // queue_name echoes the caller-provided queue name from the request, so the
+    // consumer can route the result by queue without loading state first.
+    string queue_name = 5;
 }

doc/rfc/submitqueue/extension-contract.md

@@ -14,7 +14,7 @@ Both unblock with the shape `conflict` already uses: accept identity, resolve in
 ## Principle
 
 - **Decision/action extensions** take orchestrator identity at their stage granularity and resolve granular content through narrowly-injected dependencies. Request stage → `entity.Request`; batch stage → `entity.Batch` / `[]entity.Batch`. Both are thin reference entities (a `Request` carries URIs, not diffs; a `Batch` carries IDs, not changes).
-- **Resolution targets** — `storage`, `changestore`, `queueconfig` — stay key/value-shaped. They are what the others resolve *through* (see [storage/README.md](../../../submitqueue/extension/storage/README.md) and CLAUDE.md).
+- **Resolution targets** — `storage`, `changestore`, `queueconfig` — stay key/value-shaped. They are what the others resolve *through* (see [storage/README.md](../../../submitqueue/extension/storage/README.md) and CLAUDE.md). Refinement: the storage *aggregate* has since gained the same per-queue factory resolution every other seam has — the stores it hands back remain strictly key/value, bound to their queue, while the cross-queue read-model stores stay individually-injected singletons.
 - **Output mirrors the input unit.** Each output element self-identifies with the input it corresponds to — `changeprovider`'s `ChangeInfo` carries its `URI`, `conflict`'s `Conflict` carries its `BatchID` — so a flat list suffices and the caller correlates results back to inputs without re-deriving boundaries. A *wrapper* entity (`entity.BatchChanges`) is introduced only to aggregate *up* to a coarser unit than the elements — the scorer needs batch-wide line/file totals, so the rollup earns its keep; no `RequestChanges` exists because nothing needs request-wide rollups. And when the input is a *collection* of independently-actioned units, the output groups by them: `pusher`, fed `[]entity.Batch`, returns outcomes grouped per batch, the same way `conflict` already tags each `Conflict` with its in-flight `BatchID`.
 
 ### What each stage resolves today

runway/controller/dlq/dlq.go

@@ -137,9 +137,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	)
 
 	result := &runwaymq.MergeResult{
-		Id:      request.GetId(),
-		Outcome: runwaypb.Outcome_FAILED,
-		Reason:  fmt.Sprintf("dead-lettered: %s", reason),
+		Id:        request.GetId(),
+		Outcome:   runwaypb.Outcome_FAILED,
+		Reason:    fmt.Sprintf("dead-lettered: %s", reason),
+		QueueName: request.GetQueueName(),
 	}
 
 	if err := c.publish(ctx, result, msg.PartitionKey); err != nil {

runway/controller/merge/merge.go

@@ -128,6 +128,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		}
 	}
 
+	// Echo the request's queue name so the consumer can route the result by
+	// queue without loading state first.
+	result.QueueName = request.GetQueueName()
+
 	if err := c.publish(ctx, runwaymq.TopicKeyMergeSignal, result, msg.PartitionKey); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish merge result for %s: %w", request.GetId(), err)

runway/controller/mergeconflictcheck/mergeconflictcheck.go

@@ -128,6 +128,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		}
 	}
 
+	// Echo the request's queue name so the consumer can route the result by
+	// queue without loading state first.
+	result.QueueName = request.GetQueueName()
+
 	if err := c.publish(ctx, runwaymq.TopicKeyMergeConflictCheckSignal, result, msg.PartitionKey); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish merge-conflict-check result for %s: %w", request.GetId(), err)

service/submitqueue/gateway/server/BUILD.bazel

@@ -23,8 +23,10 @@ go_library(
         "//platform/extension/messagequeue:go_default_library",
         "//platform/extension/messagequeue/mysql:go_default_library",
         "//service/submitqueue/gateway/server/mapper:go_default_library",
+        "//submitqueue/core/request:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/extension/queueconfig/yaml:go_default_library",
+        "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/extension/storage/mysql:go_default_library",
         "//submitqueue/gateway/controller:go_default_library",
         "//submitqueue/gateway/controller/log:go_default_library",

service/submitqueue/gateway/server/main.go

@@ -40,8 +40,10 @@ import (
 	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
 	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
 	"github.com/uber/submitqueue/service/submitqueue/gateway/server/mapper"
+	requestcore "github.com/uber/submitqueue/submitqueue/core/request"
 	"github.com/uber/submitqueue/submitqueue/core/topickey"
 	yamlqueueconfig "github.com/uber/submitqueue/submitqueue/extension/queueconfig/yaml"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
 	mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql"
 	"github.com/uber/submitqueue/submitqueue/gateway/controller"
 	logctrl "github.com/uber/submitqueue/submitqueue/gateway/controller/log"
@@ -321,17 +323,21 @@ func run() error {
 		return fmt.Errorf("failed to load queue configs: %w", err)
 	}
 
-	// Create controllers and wrap them for gRPC
+	// Create controllers and wrap them for gRPC. The global read-model stores
+	// are injected individually; queue-scoped storage resolves through the
+	// factory adapter, and land/cancel/log share one materializer.
+	storageFty := storageFactory{backend: store}
+	materializer := requestcore.NewMaterializer(store.GetRequestLogStore(), store.GetRequestSummaryStore(), store.GetRequestURIStore(), storageFty)
 	pingController := controller.NewPingController(logger, scope)
-	landController := controller.NewLandController(logger.Sugar(), scope, cnt, store, queueConfigs, registry)
-	cancelController := controller.NewCancelController(logger.Sugar(), scope, store, registry)
+	landController := controller.NewLandController(logger.Sugar(), scope, cnt, store.GetRequestSummaryStore(), materializer, queueConfigs, registry)
+	cancelController := controller.NewCancelController(logger.Sugar(), scope, store.GetRequestSummaryStore(), materializer, registry)
 	requestSummaryController := controller.NewRequestSummaryController(
 		logger.Sugar(),
 		scope,
 		store.GetRequestSummaryStore(),
 		store.GetRequestURIStore(),
 	)
-	listController := controller.NewListController(logger.Sugar(), scope, store.GetRequestQueueSummaryStore(), queueConfigs)
+	listController := controller.NewListController(logger.Sugar(), scope, storageFty, queueConfigs)
 	requestHistoryController := controller.NewRequestHistoryController(
 		logger.Sugar(),
 		scope,
@@ -367,7 +373,7 @@ func run() error {
 		newConsumerGate(logger),
 	)
 
-	logController := logctrl.NewController(logger.Sugar(), scope, store, topickey.TopicKeyLog, "gateway-log")
+	logController := logctrl.NewController(logger.Sugar(), scope, materializer, topickey.TopicKeyLog, "gateway-log")
 	if err := logConsumer.Register(logController); err != nil {
 		return fmt.Errorf("failed to register log controller: %w", err)
 	}
@@ -457,3 +463,16 @@ func newConsumerGate(logger *zap.Logger) consumergate.Gate {
 	logger.Info("consumer gate configured", zap.String("dir", dir))
 	return consumergatefile.New(dir)
 }
+
+// storageFactory adapts the MySQL storage backend's queue binding to the
+// storage.Factory seam. Routing every queue to the single shared backend is
+// this host's policy; a deployment that splits queues across backends swaps
+// this adapter for a routing one.
+type storageFactory struct {
+	backend *mysqlstorage.Storage
+}
+
+// For returns the queue-scoped store aggregate bound to the queue named in config.
+func (f storageFactory) For(config storage.Config) (storage.Storage, error) {
+	return f.backend.For(config.QueueName)
+}

service/submitqueue/orchestrator/server/BUILD.bazel

@@ -37,6 +37,7 @@ go_library(
         "//submitqueue/extension/conflict/fake:go_default_library",
         "//submitqueue/extension/conflict/fileoverlap:go_default_library",
         "//submitqueue/extension/conflict/none:go_default_library",
+        "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/extension/storage/mysql:go_default_library",
         "//submitqueue/extension/validator/fake:go_default_library",
         "//submitqueue/orchestrator:go_default_library",

service/submitqueue/orchestrator/server/main.go

@@ -47,6 +47,7 @@ import (
 	githubprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/github"
 	phabprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/phabricator"
 	routingprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/routing"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
 	mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql"
 	validatorfake "github.com/uber/submitqueue/submitqueue/extension/validator/fake"
 	"github.com/uber/submitqueue/submitqueue/orchestrator"
@@ -180,7 +181,8 @@ func run() error {
 	// 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.
-	profiles, err := newProfiles(logger, scope, changeset.New(store.GetRequestStore(), store.GetChangeStore()))
+	storageFty := storageFactory{backend: store}
+	profiles, err := newProfiles(logger, scope, changeset.New(storageFty))
 	if err != nil {
 		return fmt.Errorf("failed to build profiles: %w", err)
 	}
@@ -191,7 +193,7 @@ func run() error {
 	deps := orchestrator.Deps{
 		Logger:         logger.Sugar(),
 		Scope:          scope,
-		Storage:        store,
+		Storage:        storageFty,
 		Counter:        cnt,
 		BuildRunner:    profiles.BuildRunnerFactory(),
 		ChangeProvider: profiles.ChangeProviderFactory(),
@@ -431,3 +433,16 @@ func parseTimeout(envVal string, defaultVal time.Duration) time.Duration {
 	}
 	return defaultVal
 }
+
+// storageFactory adapts the MySQL storage backend's queue binding to the
+// storage.Factory seam. Routing every queue to the single shared backend is
+// this host's policy; a deployment that splits queues across backends swaps
+// this adapter for a routing one.
+type storageFactory struct {
+	backend *mysqlstorage.Storage
+}
+
+// For returns the queue-scoped store aggregate bound to the queue named in config.
+func (f storageFactory) For(config storage.Config) (storage.Storage, error) {
+	return f.backend.For(config.QueueName)
+}

submitqueue/core/batch/list.go

@@ -28,9 +28,9 @@ import (
 // ListByStates call issues while hydrating candidate IDs.
 const hydrateConcurrency = 16
 
-// ListByStates returns the queue's batches whose current state is one of the given
-// states, read through the queue's membership records: each requested state bucket
-// is listed, candidate IDs are deduplicated across buckets, every candidate is
+// ListByStates returns the bound queue's batches whose current state is one of the
+// given states, read through the queue's membership records: each requested state
+// bucket is listed, candidate IDs are deduplicated across buckets, every candidate is
 // hydrated by key with bounded concurrency, and the result keeps only batches whose
 // hydrated State is in states. Classification always uses the hydrated state — a
 // record found in a stale bucket can therefore never misreport a batch, only route
@@ -39,7 +39,7 @@ const hydrateConcurrency = 16
 // A candidate ID whose batch does not exist is returned as an error rather than
 // skipped: batch rows are never deleted, so a dangling record means the store is
 // inconsistent, not that the batch concluded.
-func ListByStates(ctx context.Context, store storage.Storage, queue string, states []entity.BatchState) ([]entity.Batch, error) {
+func ListByStates(ctx context.Context, store storage.Storage, states []entity.BatchState) ([]entity.Batch, error) {
 	wanted := make(map[entity.BatchState]bool, len(states))
 	seen := make(map[string]bool)
 	var ids []string
@@ -49,9 +49,9 @@ func ListByStates(ctx context.Context, store storage.Storage, queue string, stat
 		}
 		wanted[state] = true
 
-		records, err := store.GetQueueBatchStateStore().List(ctx, queue, state)
+		records, err := store.GetQueueBatchStateStore().List(ctx, state)
 		if err != nil {
-			return nil, fmt.Errorf("failed to list queue batch state records for queue %s state %s: %w", queue, state, err)
+			return nil, fmt.Errorf("failed to list queue batch state records for state %s: %w", state, err)
 		}
 		for _, record := range records {
 			if seen[record.BatchID] {
@@ -69,7 +69,7 @@ func ListByStates(ctx context.Context, store storage.Storage, queue string, stat
 		g.Go(func() error {
 			batch, err := store.GetBatchStore().Get(gctx, id)
 			if err != nil {
-				return fmt.Errorf("failed to get batch %s of queue %s: %w", id, queue, err)
+				return fmt.Errorf("failed to get batch %s: %w", id, err)
 			}
 			hydrated[i] = batch
 			return nil