diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go
index 062c14b7..89cf735c 100644
--- a/service/stovepipe/server/main.go
+++ b/service/stovepipe/server/main.go
@@ -249,11 +249,12 @@ func run() error {
 	scf := fakeSourceControlFactory{}
 	brf := fakeBuildRunnerFactory{}
 
-	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, store, registry, scf, brf)
+	storageFty := storageFactory{backend: store}
+	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, registry, scf, brf)
 	if err != nil {
 		return err
 	}
-	dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, store, registry)
+	dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, registry)
 	if err != nil {
 		return err
 	}
@@ -282,7 +283,7 @@ func run() error {
 		scope,
 		newInMemoryCounter(),
 		scf,
-		store,
+		storageFty,
 		registry,
 	)
 	srv := &StovepipeServer{
@@ -359,7 +360,7 @@ func registerPrimaryControllers(
 	c consumer.Consumer,
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
-	store storage.Storage,
+	store storage.Factory,
 	registry consumer.TopicRegistry,
 	scf sourcecontrol.Factory,
 	brf buildrunner.Factory,
@@ -402,7 +403,7 @@ func registerDLQControllers(
 	c consumer.Consumer,
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
-	store storage.Storage,
+	store storage.Factory,
 	registry consumer.TopicRegistry,
 ) (int, error) {
 	var count int
@@ -461,3 +462,16 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 		},
 	})
 }
+
+// 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 *storageMySQL.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)
+}
diff --git a/stovepipe/controller/build/build.go b/stovepipe/controller/build/build.go
index f5048176..8da0bc72 100644
--- a/stovepipe/controller/build/build.go
+++ b/stovepipe/controller/build/build.go
@@ -42,7 +42,7 @@ import (
 type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
-	store         storage.Storage
+	stores        storage.Factory
 	buildRunners  buildrunner.Factory
 	registry      consumer.TopicRegistry
 	topicKey      consumer.TopicKey
@@ -59,7 +59,7 @@ const _opName = "build"
 func NewController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
-	store storage.Storage,
+	stores storage.Factory,
 	buildRunners buildrunner.Factory,
 	registry consumer.TopicRegistry,
 	topicKey consumer.TopicKey,
@@ -68,7 +68,7 @@ func NewController(
 	return &Controller{
 		logger:        logger.Named("build_controller"),
 		metricsScope:  scope.SubScope("build_controller"),
-		store:         store,
+		stores:        stores,
 		buildRunners:  buildRunners,
 		registry:      registry,
 		topicKey:      topicKey,
@@ -89,12 +89,26 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to deserialize build request: %w", err)
 	}
 
-	request, err := c.loadRequest(ctx, br.Id)
+	store, err := c.stores.For(storage.Config{QueueName: br.GetQueueName()})
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, _opName, "storage_resolve_errors", 1)
+		// Non-retryable: a missing or unresolvable queue is a malformed message.
+		return fmt.Errorf("failed to resolve storage for queue %q: %w", br.GetQueueName(), err)
+	}
+
+	request, err := c.loadRequest(ctx, store, br.Id)
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
 		return err
 	}
 
+	// The payload's queue must match the request's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if br.GetQueueName() != "" && br.GetQueueName() != request.Queue {
+		metrics.NamedCounter(c.metricsScope, _opName, "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of request %s", br.GetQueueName(), request.Queue, request.ID)
+	}
+
 	// A redelivery after the build outcome was already recorded, or after process
 	// superseded the head, must not start a fresh build.
 	if request.State.IsTerminal() {
@@ -128,11 +142,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		Status:    entity.BuildStatusAccepted,
 		Version:   1,
 	}
-	if err := c.store.GetBuildStore().Create(ctx, build); err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
+	if err := store.GetBuildStore().Create(ctx, build); err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
 		return fmt.Errorf("failed to persist build %s: %w", build.ID, err)
 	}
 
-	if err := c.publishBuildSignal(ctx, build.ID); err != nil {
+	if err := c.publishBuildSignal(ctx, build.ID, request.Queue); err != nil {
 		return fmt.Errorf("failed to publish build signal for %s: %w", build.ID, err)
 	}
 
@@ -146,14 +160,14 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 }
 
 // loadRequest returns the request for id.
-func (c *Controller) loadRequest(ctx context.Context, id string) (entity.Request, error) {
-	return loader.ByID(ctx, id, c.store.GetRequestStore().Get, "request")
+func (c *Controller) loadRequest(ctx context.Context, store storage.Storage, id string) (entity.Request, error) {
+	return loader.ByID(ctx, id, store.GetRequestStore().Get, "request")
 }
 
 // publishBuildSignal publishes buildID to the buildsignal stage, partitioned by
 // build id so each build's poll loop runs in its own partition.
-func (c *Controller) publishBuildSignal(ctx context.Context, buildID string) error {
-	payload, err := stovepipemq.Marshal(&stovepipemq.BuildSignal{Id: buildID})
+func (c *Controller) publishBuildSignal(ctx context.Context, buildID, queue string) error {
+	payload, err := stovepipemq.Marshal(&stovepipemq.BuildSignal{Id: buildID, QueueName: queue})
 	if err != nil {
 		return fmt.Errorf("failed to serialize build signal: %w", err)
 	}
diff --git a/stovepipe/controller/build/build_test.go b/stovepipe/controller/build/build_test.go
index ebcbf8d9..1e7d8520 100644
--- a/stovepipe/controller/build/build_test.go
+++ b/stovepipe/controller/build/build_test.go
@@ -54,6 +54,12 @@ type buildMocks struct {
 	publisher     *mqmock.MockPublisher
 }
 
+// staticStorageFactory resolves every queue to one fixed store aggregate.
+type staticStorageFactory struct{ store storage.Storage }
+
+// For returns the fixed store aggregate for any queue.
+func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { return f.store, nil }
+
 func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildMocks) {
 	t.Helper()
 
@@ -77,7 +83,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildMoc
 	})
 	require.NoError(t, err)
 
-	c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, m.runnerFactory, registry, stovepipemq.TopicKeyBuild, "stovepipe-build")
+	c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, m.runnerFactory, registry, stovepipemq.TopicKeyBuild, "stovepipe-build")
 	return c, m
 }
 
diff --git a/stovepipe/controller/buildsignal/buildsignal.go b/stovepipe/controller/buildsignal/buildsignal.go
index fb265319..e58742c3 100644
--- a/stovepipe/controller/buildsignal/buildsignal.go
+++ b/stovepipe/controller/buildsignal/buildsignal.go
@@ -59,7 +59,7 @@ var (
 type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
-	store         storage.Storage
+	stores        storage.Factory
 	buildRunners  buildrunner.Factory
 	registry      consumer.TopicRegistry
 	topicKey      consumer.TopicKey
@@ -76,7 +76,7 @@ const _opName = "buildsignal"
 func NewController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
-	store storage.Storage,
+	stores storage.Factory,
 	buildRunners buildrunner.Factory,
 	registry consumer.TopicRegistry,
 	topicKey consumer.TopicKey,
@@ -85,7 +85,7 @@ func NewController(
 	return &Controller{
 		logger:        logger.Named("buildsignal_controller"),
 		metricsScope:  scope.SubScope("buildsignal_controller"),
-		store:         store,
+		stores:        stores,
 		buildRunners:  buildRunners,
 		registry:      registry,
 		topicKey:      topicKey,
@@ -109,18 +109,32 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to deserialize build signal: %w", err)
 	}
 
-	build, err := c.loadBuild(ctx, sig.Id)
+	store, err := c.stores.For(storage.Config{QueueName: sig.GetQueueName()})
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, _opName, "storage_resolve_errors", 1)
+		// Non-retryable: a missing or unresolvable queue is a malformed message.
+		return fmt.Errorf("failed to resolve storage for queue %q: %w", sig.GetQueueName(), err)
+	}
+
+	build, err := c.loadBuild(ctx, store, sig.Id)
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
 		return err
 	}
 
-	request, err := c.loadRequest(ctx, build.RequestID)
+	request, err := c.loadRequest(ctx, store, build.RequestID)
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
 		return err
 	}
 
+	// The payload's queue must match the request's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if sig.GetQueueName() != "" && sig.GetQueueName() != request.Queue {
+		metrics.NamedCounter(c.metricsScope, _opName, "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of request %s", sig.GetQueueName(), request.Queue, request.ID)
+	}
+
 	// A request only reaches the poll loop after process admitted it, so it is
 	// `processing` — or it already carries this build's outcome, on a redelivery
 	// after the outcome was stamped but before record was published. Both cases
@@ -147,16 +161,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to poll status for build %s: %w", build.ID, err)
 	}
 
-	effective, err := c.reconcile(ctx, build, status)
+	effective, err := c.reconcile(ctx, store, build, status)
 	if err != nil {
 		return err
 	}
 
 	if effective.IsTerminal() {
-		if err := c.finishRequest(ctx, &request, effective); err != nil {
+		if err := c.finishRequest(ctx, store, &request, effective); err != nil {
 			return err
 		}
-		if err := c.publishRecord(ctx, request.ID); err != nil {
+		if err := c.publishRecord(ctx, request.ID, request.Queue); err != nil {
 			return fmt.Errorf("failed to publish record for request %s: %w", request.ID, err)
 		}
 		c.logger.Infow("build reached terminal status",
@@ -194,17 +208,17 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 // the request non-terminal, so redelivery re-runs both steps and decrements again
 // — transiently over-admitting by one until releaseBuildSlot's zero clamp
 // reconverges, which is the failure mode this pipeline prefers.
-func (c *Controller) finishRequest(ctx context.Context, request *entity.Request, status entity.BuildStatus) error {
+func (c *Controller) finishRequest(ctx context.Context, store storage.Storage, request *entity.Request, status entity.BuildStatus) error {
 	if request.State.HasBuildOutcome() {
 		return nil
 	}
 
-	if err := c.releaseBuildSlot(ctx, request.Queue); err != nil {
+	if err := c.releaseBuildSlot(ctx, store, request.Queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
 		return err
 	}
 
-	if err := c.markOutcome(ctx, request, outcomeState(status)); err != nil {
+	if err := c.markOutcome(ctx, store, request, outcomeState(status)); err != nil {
 		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
 		return err
 	}
@@ -229,8 +243,8 @@ func outcomeState(status entity.BuildStatus) entity.RequestState {
 // conflicts. First writer wins: once any outcome is recorded a later caller leaves it
 // alone, so duplicate builds for one request (which build.md accepts) cannot flip the
 // verdict back and forth.
-func (c *Controller) markOutcome(ctx context.Context, request *entity.Request, state entity.RequestState) error {
-	reqStore := c.store.GetRequestStore()
+func (c *Controller) markOutcome(ctx context.Context, store storage.Storage, request *entity.Request, state entity.RequestState) error {
+	reqStore := store.GetRequestStore()
 
 	for {
 		if request.State != entity.RequestStateProcessing {
@@ -265,8 +279,8 @@ func (c *Controller) markOutcome(ctx context.Context, request *entity.Request, s
 // (preserving concurrent updates), clamps at zero, and retries on version conflicts.
 // Unlike process's unwind-path release this is not best-effort: the caller must not
 // mark the request terminal if the slot was not freed, so a hard failure is returned.
-func (c *Controller) releaseBuildSlot(ctx context.Context, queueName string) error {
-	queueStore := c.store.GetQueueStore()
+func (c *Controller) releaseBuildSlot(ctx context.Context, store storage.Storage, queueName string) error {
+	queueStore := store.GetQueueStore()
 
 	for {
 		queueRow, err := queueStore.Get(ctx, queueName)
@@ -295,7 +309,7 @@ func (c *Controller) releaseBuildSlot(ctx context.Context, queueName string) err
 // should drive the rest of Process: the polled status when persisted (or
 // already unchanged), or the stored status when a stored terminal status is
 // write-once-protected against a differing poll.
-func (c *Controller) reconcile(ctx context.Context, build entity.Build, status entity.BuildStatus) (entity.BuildStatus, error) {
+func (c *Controller) reconcile(ctx context.Context, store storage.Storage, build entity.Build, status entity.BuildStatus) (entity.BuildStatus, error) {
 	if status == build.Status {
 		return build.Status, nil
 	}
@@ -309,20 +323,20 @@ func (c *Controller) reconcile(ctx context.Context, build entity.Build, status e
 	newVersion := build.Version + 1
 	updated := build
 	updated.Status = status
-	if err := c.store.GetBuildStore().Update(ctx, updated, build.Version, newVersion); err != nil {
+	if err := store.GetBuildStore().Update(ctx, updated, build.Version, newVersion); err != nil {
 		return "", fmt.Errorf("failed to persist status for build %s: %w", build.ID, err)
 	}
 	return status, nil
 }
 
 // loadBuild returns the build for id.
-func (c *Controller) loadBuild(ctx context.Context, id string) (entity.Build, error) {
-	return loader.ByID(ctx, id, c.store.GetBuildStore().Get, "build")
+func (c *Controller) loadBuild(ctx context.Context, store storage.Storage, id string) (entity.Build, error) {
+	return loader.ByID(ctx, id, store.GetBuildStore().Get, "build")
 }
 
 // loadRequest returns the request for id.
-func (c *Controller) loadRequest(ctx context.Context, id string) (entity.Request, error) {
-	return loader.ByID(ctx, id, c.store.GetRequestStore().Get, "request")
+func (c *Controller) loadRequest(ctx context.Context, store storage.Storage, id string) (entity.Request, error) {
+	return loader.ByID(ctx, id, store.GetRequestStore().Get, "request")
 }
 
 // pollDelay returns the delay before the next Status call for a non-terminal status.
@@ -340,8 +354,8 @@ func pollDelay(status entity.BuildStatus) int64 {
 // id. The message id is the request id too, so a redelivery republishing the
 // same terminal signal dedups into the original message rather than enqueuing a
 // second one.
-func (c *Controller) publishRecord(ctx context.Context, requestID string) error {
-	payload, err := stovepipemq.Marshal(&stovepipemq.Record{Id: requestID})
+func (c *Controller) publishRecord(ctx context.Context, requestID, queue string) error {
+	payload, err := stovepipemq.Marshal(&stovepipemq.Record{Id: requestID, QueueName: queue})
 	if err != nil {
 		return fmt.Errorf("failed to serialize record: %w", err)
 	}
diff --git a/stovepipe/controller/buildsignal/buildsignal_test.go b/stovepipe/controller/buildsignal/buildsignal_test.go
index 53daa39d..022ff927 100644
--- a/stovepipe/controller/buildsignal/buildsignal_test.go
+++ b/stovepipe/controller/buildsignal/buildsignal_test.go
@@ -54,6 +54,12 @@ type buildsignalMocks struct {
 	publisher     *mqmock.MockPublisher
 }
 
+// staticStorageFactory resolves every queue to one fixed store aggregate.
+type staticStorageFactory struct{ store storage.Storage }
+
+// For returns the fixed store aggregate for any queue.
+func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { return f.store, nil }
+
 func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildsignalMocks) {
 	t.Helper()
 
@@ -80,7 +86,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildsig
 	})
 	require.NoError(t, err)
 
-	c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, m.runnerFactory, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal")
+	c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, m.runnerFactory, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal")
 	return c, m
 }
 
@@ -423,7 +429,7 @@ func TestPublishRecordCarriesRequestID(t *testing.T) {
 			return nil
 		})
 
-	require.NoError(t, c.publishRecord(context.Background(), testID))
+	require.NoError(t, c.publishRecord(context.Background(), testID, "monorepo/main"))
 
 	var payload stovepipemq.Record
 	require.NoError(t, stovepipemq.Unmarshal(got.Payload, &payload))
diff --git a/stovepipe/controller/dlq/dlq_test.go b/stovepipe/controller/dlq/dlq_test.go
index 010ca40a..93e1503e 100644
--- a/stovepipe/controller/dlq/dlq_test.go
+++ b/stovepipe/controller/dlq/dlq_test.go
@@ -42,6 +42,12 @@ type dlqMocks struct {
 	queueStore *storagemock.MockQueueStore
 }
 
+// staticStorageFactory resolves every queue to one fixed store aggregate.
+type staticStorageFactory struct{ store storage.Storage }
+
+// For returns the fixed store aggregate for any queue.
+func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { return f.store, nil }
+
 func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, dlqMocks) {
 	t.Helper()
 
@@ -54,7 +60,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, dlqMocks
 	store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes()
 	store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes()
 
-	c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq")
+	c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq")
 	return c, m
 }
 
diff --git a/stovepipe/controller/dlq/request.go b/stovepipe/controller/dlq/request.go
index 262fd996..a58fefb7 100644
--- a/stovepipe/controller/dlq/request.go
+++ b/stovepipe/controller/dlq/request.go
@@ -33,7 +33,7 @@ import (
 type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
-	store         storage.Storage
+	stores        storage.Factory
 	topicKey      consumer.TopicKey
 	consumerGroup string
 }
@@ -49,14 +49,14 @@ const _opName = "process_dlq"
 func NewController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
-	store storage.Storage,
+	stores storage.Factory,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
 ) *Controller {
 	return &Controller{
 		logger:        logger.Named("process_dlq_controller"),
 		metricsScope:  scope.SubScope("process_dlq_controller"),
-		store:         store,
+		stores:        stores,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
 	}
@@ -88,6 +88,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("dlq payload decoded to empty request id")
 	}
 
+	store, err := c.stores.For(storage.Config{QueueName: pr.GetQueueName()})
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, _opName, "storage_resolve_errors", 1)
+		// Non-retryable: a missing or unresolvable queue is a malformed message.
+		return fmt.Errorf("failed to resolve storage for queue %q: %w", pr.GetQueueName(), err)
+	}
+
 	dmeta := delivery.Metadata()
 	c.logger.Warnw("dlq message received",
 		"request_id", pr.Id,
@@ -97,7 +104,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		"dlq_last_error", dmeta["dlq.last_error"],
 	)
 
-	if err := failRequest(ctx, c.store, c.logger, pr.Id); err != nil {
+	if err := failRequest(ctx, store, c.logger, pr.Id); err != nil {
 		metrics.NamedCounter(c.metricsScope, _opName, "reconcile_errors", 1)
 		return err
 	}
diff --git a/stovepipe/controller/ingest.go b/stovepipe/controller/ingest.go
index 729420eb..a9253b0f 100644
--- a/stovepipe/controller/ingest.go
+++ b/stovepipe/controller/ingest.go
@@ -53,7 +53,7 @@ type IngestController struct {
 	metricsScope  tally.Scope
 	counter       counter.Counter
 	sourceControl sourcecontrol.Factory
-	store         storage.Storage
+	stores        storage.Factory
 	registry      consumer.TopicRegistry
 }
 
@@ -64,7 +64,7 @@ func NewIngestController(
 	scope tally.Scope,
 	counter counter.Counter,
 	sourceControl sourcecontrol.Factory,
-	store storage.Storage,
+	stores storage.Factory,
 	registry consumer.TopicRegistry,
 ) *IngestController {
 	return &IngestController{
@@ -72,7 +72,7 @@ func NewIngestController(
 		metricsScope:  scope.SubScope("ingest_controller"),
 		counter:       counter,
 		sourceControl: sourceControl,
-		store:         store,
+		stores:        stores,
 		registry:      registry,
 	}
 }
@@ -97,6 +97,11 @@ func (c *IngestController) Ingest(ctx context.Context, req entity.IngestRequest)
 	}
 	queue := req.Queue
 
+	store, err := c.stores.For(storage.Config{QueueName: queue})
+	if err != nil {
+		return entity.IngestResult{}, fmt.Errorf("failed to resolve storage for queue %q: %w", queue, err)
+	}
+
 	// Resolve the queue's current head commit to its opaque URI via SourceControl.
 	// An unresolvable queue/ref is a caller error (unknown queue), not infrastructure.
 	sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: queue})
@@ -113,19 +118,19 @@ func (c *IngestController) Ingest(ctx context.Context, req entity.IngestRequest)
 
 	// The (queue, URI) mapping is the dedup gate and the source of truth for "does this head
 	// have a request id".
-	id, err := c.resolveID(ctx, queue, uri)
+	id, err := c.resolveID(ctx, store, queue, uri)
 	if err != nil {
 		return entity.IngestResult{}, err
 	}
 
 	// Ensure the request row exists, healing a prior partial write where the mapping committed
 	// but the request did not.
-	request, err := c.ensureRequest(ctx, id, queue, uri)
+	request, err := c.ensureRequest(ctx, store, id, queue, uri)
 	if err != nil {
 		return entity.IngestResult{}, err
 	}
 
-	if err := c.advanceQueueLatestRequestID(ctx, queue, id); err != nil {
+	if err := c.advanceQueueLatestRequestID(ctx, store, queue, id); err != nil {
 		return entity.IngestResult{}, err
 	}
 
@@ -153,10 +158,10 @@ func (c *IngestController) Ingest(ctx context.Context, req entity.IngestRequest)
 // pair is not yet mapped. Claiming the mapping is the dedup gate: a concurrent ingest that loses
 // the claim re-reads and returns the winner's id, so no orphan request row is created (only the
 // minted counter value is spent).
-func (c *IngestController) resolveID(ctx context.Context, queue, uri string) (string, error) {
-	uriStore := c.store.GetRequestURIStore()
+func (c *IngestController) resolveID(ctx context.Context, store storage.Storage, queue, uri string) (string, error) {
+	uriStore := store.GetRequestURIStore()
 
-	if id, err := uriStore.GetIDByURI(ctx, queue, uri); err == nil {
+	if id, err := uriStore.GetIDByURI(ctx, uri); err == nil {
 		return id, nil
 	} else if !errors.Is(err, storage.ErrNotFound) {
 		return "", fmt.Errorf("failed to look up existing request for queue=%s: %w", queue, err)
@@ -171,9 +176,9 @@ func (c *IngestController) resolveID(ctx context.Context, queue, uri string) (st
 	}
 	id := fmt.Sprintf("%s/%d", domain, seq)
 
-	if err := uriStore.Create(ctx, queue, uri, id); err != nil {
+	if err := uriStore.Create(ctx, uri, id); err != nil {
 		if errors.Is(err, storage.ErrAlreadyExists) {
-			existing, getErr := uriStore.GetIDByURI(ctx, queue, uri)
+			existing, getErr := uriStore.GetIDByURI(ctx, uri)
 			if getErr != nil {
 				return "", fmt.Errorf("failed to resolve raced request for queue=%s: %w", queue, getErr)
 			}
@@ -186,8 +191,8 @@ func (c *IngestController) resolveID(ctx context.Context, queue, uri string) (st
 
 // ensureRequest returns the request for id, creating it in the Accepted state if it does not yet
 // exist. A concurrent creator (ErrAlreadyExists) is resolved by re-reading the canonical row.
-func (c *IngestController) ensureRequest(ctx context.Context, id, queue, uri string) (entity.Request, error) {
-	reqStore := c.store.GetRequestStore()
+func (c *IngestController) ensureRequest(ctx context.Context, store storage.Storage, id, queue, uri string) (entity.Request, error) {
+	reqStore := store.GetRequestStore()
 
 	got, err := reqStore.Get(ctx, id)
 	if err == nil {
@@ -216,8 +221,8 @@ func (c *IngestController) ensureRequest(ctx context.Context, id, queue, uri str
 
 // ensureQueue returns the queue row for name, creating it if it does not yet exist.
 // A concurrent creator (ErrAlreadyExists) is resolved by re-reading the canonical row.
-func (c *IngestController) ensureQueue(ctx context.Context, name string) (entity.Queue, error) {
-	queueStore := c.store.GetQueueStore()
+func (c *IngestController) ensureQueue(ctx context.Context, store storage.Storage, name string) (entity.Queue, error) {
+	queueStore := store.GetQueueStore()
 
 	got, err := queueStore.Get(ctx, name)
 	if err == nil {
@@ -243,11 +248,11 @@ func (c *IngestController) ensureQueue(ctx context.Context, name string) (entity
 
 // advanceQueueLatestRequestID CAS-updates queue.latest_request_id to id when id is newer.
 // Retries on optimistic-lock conflicts so concurrent ingests converge.
-func (c *IngestController) advanceQueueLatestRequestID(ctx context.Context, queue, id string) error {
-	queueStore := c.store.GetQueueStore()
+func (c *IngestController) advanceQueueLatestRequestID(ctx context.Context, store storage.Storage, queue, id string) error {
+	queueStore := store.GetQueueStore()
 
 	for {
-		queueRow, err := c.ensureQueue(ctx, queue)
+		queueRow, err := c.ensureQueue(ctx, store, queue)
 		if err != nil {
 			return err
 		}
@@ -277,7 +282,7 @@ func (c *IngestController) advanceQueueLatestRequestID(ctx context.Context, queu
 // publishProcess publishes the request ID to the process stage, partitioned by queue so a
 // queue's requests stay ordered.
 func (c *IngestController) publishProcess(ctx context.Context, id, queue string) error {
-	payload, err := stovepipemq.Marshal(&stovepipemq.ProcessRequest{Id: id})
+	payload, err := stovepipemq.Marshal(&stovepipemq.ProcessRequest{Id: id, QueueName: queue})
 	if err != nil {
 		return fmt.Errorf("failed to serialize process request: %w", err)
 	}
diff --git a/stovepipe/controller/ingest_test.go b/stovepipe/controller/ingest_test.go
index 18cd66f2..af8cec37 100644
--- a/stovepipe/controller/ingest_test.go
+++ b/stovepipe/controller/ingest_test.go
@@ -51,6 +51,12 @@ type ingestMocks struct {
 	publisher  *mqmock.MockPublisher
 }
 
+// staticStorageFactory resolves every queue to one fixed store aggregate.
+type staticStorageFactory struct{ store storage.Storage }
+
+// For returns the fixed store aggregate for any queue.
+func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { return f.store, nil }
+
 func newIngestController(t *testing.T, ctrl *gomock.Controller) (*IngestController, ingestMocks) {
 	t.Helper()
 
@@ -77,7 +83,7 @@ func newIngestController(t *testing.T, ctrl *gomock.Controller) (*IngestControll
 	})
 	require.NoError(t, err)
 
-	c := NewIngestController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), m.counter, m.factory, store, registry)
+	c := NewIngestController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), m.counter, m.factory, staticStorageFactory{store: store}, registry)
 	return c, m
 }
 
@@ -118,9 +124,9 @@ func TestIngestController_Ingest(t *testing.T) {
 			queue: testQueue,
 			setup: func(m ingestMocks) {
 				expectResolve(m)
-				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("", storage.ErrNotFound)
+				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("", storage.ErrNotFound)
 				m.counter.EXPECT().Next(gomock.Any(), "request/"+testQueue).Return(int64(7), nil)
-				m.uriStore.EXPECT().Create(gomock.Any(), testQueue, testURI, "request/monorepo/main/7").Return(nil)
+				m.uriStore.EXPECT().Create(gomock.Any(), testURI, "request/monorepo/main/7").Return(nil)
 				m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/7").Return(entity.Request{}, storage.ErrNotFound)
 				m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
 				expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/7")
@@ -133,7 +139,7 @@ func TestIngestController_Ingest(t *testing.T) {
 			queue: testQueue,
 			setup: func(m ingestMocks) {
 				expectResolve(m)
-				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("request/monorepo/main/3", nil)
+				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("request/monorepo/main/3", nil)
 				m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{ID: "request/monorepo/main/3", State: entity.RequestStateAccepted}, nil)
 				expectAdvanceLatestRequestIDNoOp(m, testQueue, "request/monorepo/main/3")
 				m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
@@ -145,7 +151,7 @@ func TestIngestController_Ingest(t *testing.T) {
 			queue: testQueue,
 			setup: func(m ingestMocks) {
 				expectResolve(m)
-				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("request/monorepo/main/3", nil)
+				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("request/monorepo/main/3", nil)
 				m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{}, storage.ErrNotFound)
 				m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
 				expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/3")
@@ -158,10 +164,10 @@ func TestIngestController_Ingest(t *testing.T) {
 			queue: testQueue,
 			setup: func(m ingestMocks) {
 				expectResolve(m)
-				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("", storage.ErrNotFound)
+				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("", storage.ErrNotFound)
 				m.counter.EXPECT().Next(gomock.Any(), "request/"+testQueue).Return(int64(7), nil)
-				m.uriStore.EXPECT().Create(gomock.Any(), testQueue, testURI, "request/monorepo/main/7").Return(storage.ErrAlreadyExists)
-				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("request/monorepo/main/3", nil)
+				m.uriStore.EXPECT().Create(gomock.Any(), testURI, "request/monorepo/main/7").Return(storage.ErrAlreadyExists)
+				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("request/monorepo/main/3", nil)
 				m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{ID: "request/monorepo/main/3", State: entity.RequestStateAccepted}, nil)
 				expectAdvanceLatestRequestIDNoOp(m, testQueue, "request/monorepo/main/3")
 				m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
@@ -199,7 +205,7 @@ func TestIngestController_Ingest(t *testing.T) {
 			queue: testQueue,
 			setup: func(m ingestMocks) {
 				expectResolve(m)
-				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("", storage.ErrNotFound)
+				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("", storage.ErrNotFound)
 				m.counter.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(0), errors.New("counter unavailable"))
 			},
 			wantErr: true,
@@ -209,9 +215,9 @@ func TestIngestController_Ingest(t *testing.T) {
 			queue: testQueue,
 			setup: func(m ingestMocks) {
 				expectResolve(m)
-				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("", storage.ErrNotFound)
+				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("", storage.ErrNotFound)
 				m.counter.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(7), nil)
-				m.uriStore.EXPECT().Create(gomock.Any(), testQueue, testURI, gomock.Any()).Return(nil)
+				m.uriStore.EXPECT().Create(gomock.Any(), testURI, gomock.Any()).Return(nil)
 				m.reqStore.EXPECT().Get(gomock.Any(), gomock.Any()).Return(entity.Request{}, storage.ErrNotFound)
 				m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(errors.New("db down"))
 			},
@@ -222,9 +228,9 @@ func TestIngestController_Ingest(t *testing.T) {
 			queue: testQueue,
 			setup: func(m ingestMocks) {
 				expectResolve(m)
-				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("", storage.ErrNotFound)
+				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("", storage.ErrNotFound)
 				m.counter.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(7), nil)
-				m.uriStore.EXPECT().Create(gomock.Any(), testQueue, testURI, gomock.Any()).Return(nil)
+				m.uriStore.EXPECT().Create(gomock.Any(), testURI, gomock.Any()).Return(nil)
 				m.reqStore.EXPECT().Get(gomock.Any(), gomock.Any()).Return(entity.Request{}, storage.ErrNotFound)
 				m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
 				expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/7")
diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go
index f02b864e..d4f36bef 100644
--- a/stovepipe/controller/process/process.go
+++ b/stovepipe/controller/process/process.go
@@ -43,7 +43,7 @@ import (
 type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
-	store         storage.Storage
+	stores        storage.Factory
 	queueConfigs  queueconfig.Store
 	sourceControl sourcecontrol.Factory
 	registry      consumer.TopicRegistry
@@ -61,7 +61,7 @@ const _opName = "process"
 func NewController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
-	store storage.Storage,
+	stores storage.Factory,
 	queueConfigs queueconfig.Store,
 	sourceControl sourcecontrol.Factory,
 	registry consumer.TopicRegistry,
@@ -71,7 +71,7 @@ func NewController(
 	return &Controller{
 		logger:        logger.Named("process_controller"),
 		metricsScope:  scope.SubScope("process_controller"),
-		store:         store,
+		stores:        stores,
 		queueConfigs:  queueConfigs,
 		sourceControl: sourceControl,
 		registry:      registry,
@@ -92,15 +92,29 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to deserialize process request: %w", err)
 	}
 
-	request, err := c.loadRequest(ctx, pr.Id)
+	store, err := c.stores.For(storage.Config{QueueName: pr.GetQueueName()})
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, _opName, "storage_resolve_errors", 1)
+		// Non-retryable: a missing or unresolvable queue is a malformed message.
+		return fmt.Errorf("failed to resolve storage for queue %q: %w", pr.GetQueueName(), err)
+	}
+
+	request, err := c.loadRequest(ctx, store, pr.Id)
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
 		return err
 	}
 
+	// The payload's queue must match the request's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if pr.GetQueueName() != "" && pr.GetQueueName() != request.Queue {
+		metrics.NamedCounter(c.metricsScope, _opName, "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of request %s", pr.GetQueueName(), request.Queue, request.ID)
+	}
+
 	switch request.State {
 	case entity.RequestStateProcessing:
-		if err := c.publishBuild(ctx, request.ID); err != nil {
+		if err := c.publishBuild(ctx, request.ID, request.Queue); err != nil {
 			metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1)
 			return fmt.Errorf("failed to publish request %s to build: %w", request.ID, err)
 		}
@@ -110,7 +124,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		// A stale redelivery has nothing left to do.
 		return nil
 	case entity.RequestStateAccepted:
-		return c.processAccepted(ctx, delivery, request)
+		return c.processAccepted(ctx, store, delivery, request)
 	default:
 		c.logger.Warnw("ignored request in unexpected state",
 			"request_id", request.ID,
@@ -124,8 +138,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 // processAccepted coalesces older heads against queue.latest_request_id, then admits
 // the latest head when a build slot is available. The delivery is threaded down so a
 // closed gate can hold it.
-func (c *Controller) processAccepted(ctx context.Context, delivery consumer.Delivery, request entity.Request) error {
-	queueRow, err := c.loadQueue(ctx, request.Queue)
+func (c *Controller) processAccepted(ctx context.Context, store storage.Storage, delivery consumer.Delivery, request entity.Request) error {
+	queueRow, err := c.loadQueue(ctx, store, request.Queue)
 	if err != nil {
 		if !errs.IsRetryable(err) {
 			metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
@@ -142,7 +156,7 @@ func (c *Controller) processAccepted(ctx context.Context, delivery consumer.Deli
 		return nil
 	}
 
-	superseded, err := c.coalesce(ctx, request, queueRow.LatestRequestID)
+	superseded, err := c.coalesce(ctx, store, request, queueRow.LatestRequestID)
 	if err != nil || superseded {
 		return err
 	}
@@ -154,13 +168,13 @@ func (c *Controller) processAccepted(ctx context.Context, delivery consumer.Deli
 		return fmt.Errorf("failed to load queue config for %s: %w", request.Queue, err)
 	}
 
-	return c.admitLatestHead(ctx, delivery, request, queueRow, cfg)
+	return c.admitLatestHead(ctx, store, delivery, request, queueRow, cfg)
 }
 
 // coalesce supersedes request when a newer head exists (RFC process step 5), returning
 // true so the caller acks. It returns false when request is still the latest head and
 // should proceed to the gate. Superseding consumes no build slot.
-func (c *Controller) coalesce(ctx context.Context, request entity.Request, latestRequestID string) (bool, error) {
+func (c *Controller) coalesce(ctx context.Context, store storage.Storage, request entity.Request, latestRequestID string) (bool, error) {
 	cmp, err := entity.CompareRequestID(request.Queue, request.ID, latestRequestID)
 	if err != nil {
 		return false, fmt.Errorf("failed to compare request ids for queue %s: %w", request.Queue, err)
@@ -168,7 +182,7 @@ func (c *Controller) coalesce(ctx context.Context, request entity.Request, lates
 	if cmp >= 0 {
 		return false, nil
 	}
-	if err := c.supersedeRequest(ctx, request); err != nil {
+	if err := c.supersedeRequest(ctx, store, request); err != nil {
 		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
 		return false, err
 	}
@@ -185,7 +199,7 @@ func (c *Controller) coalesce(ctx context.Context, request entity.Request, lates
 // slot, mark the request processing, and publish it to build. Every queue-row reload
 // re-runs coalesce-then-gate, so a slot is never spent on a now-stale head; a closed gate
 // defers by holding the delivery (redeliver after the gate wait delay) rather than failing.
-func (c *Controller) admitLatestHead(ctx context.Context, delivery consumer.Delivery, request entity.Request, queueRow entity.Queue, cfg entity.QueueConfig) error {
+func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage, delivery consumer.Delivery, request entity.Request, queueRow entity.Queue, cfg entity.QueueConfig) error {
 	var sc sourcecontrol.SourceControl
 	var strategy entity.BuildStrategy
 	var baseURI string
@@ -211,7 +225,7 @@ func (c *Controller) admitLatestHead(ctx context.Context, delivery consumer.Deli
 			return err
 		}
 
-		err = c.claimBuildSlot(ctx, &queueRow)
+		err = c.claimBuildSlot(ctx, store, &queueRow)
 		if err == nil {
 			break
 		}
@@ -220,26 +234,26 @@ func (c *Controller) admitLatestHead(ctx context.Context, delivery consumer.Deli
 		}
 		// claimBuildSlot reloaded queueRow. Re-coalesce: supersede if a newer head arrived,
 		// otherwise loop to re-check the gate.
-		superseded, err := c.coalesce(ctx, request, queueRow.LatestRequestID)
+		superseded, err := c.coalesce(ctx, store, request, queueRow.LatestRequestID)
 		if err != nil || superseded {
 			return err
 		}
 	}
 
-	transitioned, err := c.markProcessing(ctx, &request, strategy, baseURI)
+	transitioned, err := c.markProcessing(ctx, store, &request, strategy, baseURI)
 	if err != nil {
 		// Slot claimed but never admitted: release best-effort so the slot isn't leaked
 		// (a redelivery would find the gate closed by its own claim and nothing decrements it).
-		c.releaseBuildSlot(ctx, request.Queue)
+		c.releaseBuildSlot(ctx, store, request.Queue)
 		return err
 	}
 	if !transitioned {
 		// Lost the admit race: another delivery advanced this request. Release and skip.
-		c.releaseBuildSlot(ctx, request.Queue)
+		c.releaseBuildSlot(ctx, store, request.Queue)
 		return nil
 	}
 
-	if err := c.publishBuild(ctx, request.ID); err != nil {
+	if err := c.publishBuild(ctx, request.ID, request.Queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish request %s to build: %w", request.ID, err)
 	}
@@ -291,8 +305,8 @@ func (c *Controller) deriveBuildStrategy(ctx context.Context, sc sourcecontrol.S
 
 // claimBuildSlot CAS-increments queue.in_flight_count by one. On version mismatch it
 // reloads queueRow and returns ErrVersionMismatch so the caller can retry.
-func (c *Controller) claimBuildSlot(ctx context.Context, queueRow *entity.Queue) error {
-	queueStore := c.store.GetQueueStore()
+func (c *Controller) claimBuildSlot(ctx context.Context, store storage.Storage, queueRow *entity.Queue) error {
+	queueStore := store.GetQueueStore()
 
 	updated := *queueRow
 	updated.InFlightCount = queueRow.InFlightCount + 1
@@ -318,8 +332,8 @@ func (c *Controller) claimBuildSlot(ctx context.Context, queueRow *entity.Queue)
 // update cannot discard them. transitioned is true only when this call performed the CAS; false
 // means a concurrent writer already advanced the request past accepted, so the caller must release
 // its claimed slot.
-func (c *Controller) markProcessing(ctx context.Context, request *entity.Request, strategy entity.BuildStrategy, baseURI string) (transitioned bool, err error) {
-	reqStore := c.store.GetRequestStore()
+func (c *Controller) markProcessing(ctx context.Context, store storage.Storage, request *entity.Request, strategy entity.BuildStrategy, baseURI string) (transitioned bool, err error) {
+	reqStore := store.GetRequestStore()
 
 	for {
 		if request.State != entity.RequestStateAccepted {
@@ -351,8 +365,8 @@ func (c *Controller) markProcessing(ctx context.Context, request *entity.Request
 // releaseBuildSlot CAS-decrements queue.in_flight_count to compensate a slot claimed but never
 // admitted. It decrements relatively (preserving a concurrent record decrement) and retries on
 // version conflicts. Best-effort: it only logs on a hard failure, since the caller is unwinding.
-func (c *Controller) releaseBuildSlot(ctx context.Context, queueName string) {
-	queueStore := c.store.GetQueueStore()
+func (c *Controller) releaseBuildSlot(ctx context.Context, store storage.Storage, queueName string) {
+	queueStore := store.GetQueueStore()
 
 	for {
 		queueRow, err := queueStore.Get(ctx, queueName)
@@ -386,8 +400,8 @@ func (c *Controller) releaseBuildSlot(ctx context.Context, queueName string) {
 }
 
 // supersedeRequest transitions a request from accepted to superseded, retrying on version conflicts.
-func (c *Controller) supersedeRequest(ctx context.Context, request entity.Request) error {
-	reqStore := c.store.GetRequestStore()
+func (c *Controller) supersedeRequest(ctx context.Context, store storage.Storage, request entity.Request) error {
+	reqStore := store.GetRequestStore()
 
 	for {
 		if request.State != entity.RequestStateAccepted {
@@ -434,20 +448,20 @@ func (c *Controller) holdForBuildSlot(delivery consumer.Delivery, request entity
 }
 
 // loadRequest returns the request for id.
-func (c *Controller) loadRequest(ctx context.Context, id string) (entity.Request, error) {
-	return loader.ByID(ctx, id, c.store.GetRequestStore().Get, "request")
+func (c *Controller) loadRequest(ctx context.Context, store storage.Storage, id string) (entity.Request, error) {
+	return loader.ByID(ctx, id, store.GetRequestStore().Get, "request")
 }
 
 // loadQueue returns the queue row for name.
-func (c *Controller) loadQueue(ctx context.Context, name string) (entity.Queue, error) {
-	return loader.ByID(ctx, name, c.store.GetQueueStore().Get, "queue")
+func (c *Controller) loadQueue(ctx context.Context, store storage.Storage, name string) (entity.Queue, error) {
+	return loader.ByID(ctx, name, store.GetQueueStore().Get, "queue")
 }
 
 // publishBuild publishes the admitted request ID to the build stage. The build
 // controller reloads the Request from storage to read its immutable strategy
 // and baseline.
-func (c *Controller) publishBuild(ctx context.Context, id string) error {
-	payload, err := stovepipemq.Marshal(&stovepipemq.BuildRequest{Id: id})
+func (c *Controller) publishBuild(ctx context.Context, id, queue string) error {
+	payload, err := stovepipemq.Marshal(&stovepipemq.BuildRequest{Id: id, QueueName: queue})
 	if err != nil {
 		return fmt.Errorf("failed to serialize build request: %w", err)
 	}
diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go
index 770dbee4..71aae9a0 100644
--- a/stovepipe/controller/process/process_test.go
+++ b/stovepipe/controller/process/process_test.go
@@ -53,6 +53,12 @@ type processMocks struct {
 	publisher     *mqmock.MockPublisher
 }
 
+// staticStorageFactory resolves every queue to one fixed store aggregate.
+type staticStorageFactory struct{ store storage.Storage }
+
+// For returns the fixed store aggregate for any queue.
+func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { return f.store, nil }
+
 func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processMocks) {
 	t.Helper()
 	return newControllerWithScope(t, ctrl, tally.NewTestScope("test", nil))
@@ -83,7 +89,7 @@ func newControllerWithScope(t *testing.T, ctrl *gomock.Controller, scope tally.S
 	c := NewController(
 		zap.NewNop().Sugar(),
 		scope,
-		store,
+		staticStorageFactory{store: store},
 		queueconfigdefault.NewStore(),
 		m.sourceFactory,
 		registry,
diff --git a/stovepipe/core/messagequeue/proto/build.proto b/stovepipe/core/messagequeue/proto/build.proto
index b30ce26e..5b9b03bd 100644
--- a/stovepipe/core/messagequeue/proto/build.proto
+++ b/stovepipe/core/messagequeue/proto/build.proto
@@ -33,4 +33,8 @@ message BuildRequest {
 
     // id is the request id to build. Format: "request/<queue>/<counter>".
     string id = 1;
+    // queue_name is the name of the queue processing the request, carried so
+    // the consumer can route by queue without loading state first. Empty on
+    // payloads written before the field existed.
+    string queue_name = 2;
 }
diff --git a/stovepipe/core/messagequeue/proto/buildsignal.proto b/stovepipe/core/messagequeue/proto/buildsignal.proto
index 3868ad9e..9d85ff71 100644
--- a/stovepipe/core/messagequeue/proto/buildsignal.proto
+++ b/stovepipe/core/messagequeue/proto/buildsignal.proto
@@ -34,4 +34,8 @@ message BuildSignal {
     // id is the build id to poll: the runner-assigned id minted by
     // BuildRunner.Trigger (entity.Build.ID / entity.BuildID.ID).
     string id = 1;
+    // queue_name is the name of the queue processing the batch's request,
+    // carried so the consumer can route by queue without loading state first.
+    // Empty on payloads written before the field existed.
+    string queue_name = 2;
 }
diff --git a/stovepipe/core/messagequeue/proto/process.proto b/stovepipe/core/messagequeue/proto/process.proto
index 999e149c..bb0059c7 100644
--- a/stovepipe/core/messagequeue/proto/process.proto
+++ b/stovepipe/core/messagequeue/proto/process.proto
@@ -32,4 +32,8 @@ message ProcessRequest {
 
     // id is the minted request id to process. Format: "request/<queue>/<counter>".
     string id = 1;
+    // queue_name is the name of the queue processing the request, carried so
+    // the consumer can route by queue without loading state first. Empty on
+    // payloads written before the field existed.
+    string queue_name = 2;
 }
diff --git a/stovepipe/core/messagequeue/proto/record.proto b/stovepipe/core/messagequeue/proto/record.proto
index 68204ccb..dffabaf8 100644
--- a/stovepipe/core/messagequeue/proto/record.proto
+++ b/stovepipe/core/messagequeue/proto/record.proto
@@ -35,4 +35,8 @@ message Record {
     // id is the request id whose build reached a terminal status. Format:
     // "request/<queue>/<counter>" (entity.Request.ID).
     string id = 1;
+    // queue_name is the name of the queue processing the request, carried so
+    // the consumer can route by queue without loading state first. Empty on
+    // payloads written before the field existed.
+    string queue_name = 2;
 }
diff --git a/stovepipe/core/messagequeue/protopb/build.pb.go b/stovepipe/core/messagequeue/protopb/build.pb.go
index e8150153..940f0cfa 100644
--- a/stovepipe/core/messagequeue/protopb/build.pb.go
+++ b/stovepipe/core/messagequeue/protopb/build.pb.go
@@ -45,7 +45,11 @@ const (
 type BuildRequest struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// id is the request id to build. Format: "request/<queue>/<counter>".
-	Id            string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
+	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
+	// queue_name is the name of the queue processing the request, carried so
+	// the consumer can route by queue without loading state first. Empty on
+	// payloads written before the field existed.
+	QueueName     string `protobuf:"bytes,2,opt,name=queue_name,json=queueName,proto3" json:"queue_name,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -87,13 +91,22 @@ func (x *BuildRequest) GetId() string {
 	return ""
 }
 
+func (x *BuildRequest) GetQueueName() string {
+	if x != nil {
+		return x.QueueName
+	}
+	return ""
+}
+
 var File_build_proto protoreflect.FileDescriptor
 
 const file_build_proto_rawDesc = "" +
 	"\n" +
-	"\vbuild.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\")\n" +
+	"\vbuild.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"H\n" +
 	"\fBuildRequest\x12\x0e\n" +
-	"\x02id\x18\x01 \x01(\tR\x02id:\t\x8a\xb5\x18\x05buildB|\n" +
+	"\x02id\x18\x01 \x01(\tR\x02id\x12\x1d\n" +
+	"\n" +
+	"queue_name\x18\x02 \x01(\tR\tqueueName:\t\x8a\xb5\x18\x05buildB|\n" +
 	"+com.uber.submitqueue.stovepipe.messagequeueB\n" +
 	"BuildProtoP\x01Z?github.com/uber/submitqueue/stovepipe/core/messagequeue/protopbb\x06proto3"
 
diff --git a/stovepipe/core/messagequeue/protopb/buildsignal.pb.go b/stovepipe/core/messagequeue/protopb/buildsignal.pb.go
index 2e4eff18..c30d7b1d 100644
--- a/stovepipe/core/messagequeue/protopb/buildsignal.pb.go
+++ b/stovepipe/core/messagequeue/protopb/buildsignal.pb.go
@@ -46,7 +46,11 @@ type BuildSignal struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// id is the build id to poll: the runner-assigned id minted by
 	// BuildRunner.Trigger (entity.Build.ID / entity.BuildID.ID).
-	Id            string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
+	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
+	// queue_name is the name of the queue processing the batch's request,
+	// carried so the consumer can route by queue without loading state first.
+	// Empty on payloads written before the field existed.
+	QueueName     string `protobuf:"bytes,2,opt,name=queue_name,json=queueName,proto3" json:"queue_name,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -88,13 +92,22 @@ func (x *BuildSignal) GetId() string {
 	return ""
 }
 
+func (x *BuildSignal) GetQueueName() string {
+	if x != nil {
+		return x.QueueName
+	}
+	return ""
+}
+
 var File_buildsignal_proto protoreflect.FileDescriptor
 
 const file_buildsignal_proto_rawDesc = "" +
 	"\n" +
-	"\x11buildsignal.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\".\n" +
+	"\x11buildsignal.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"M\n" +
 	"\vBuildSignal\x12\x0e\n" +
-	"\x02id\x18\x01 \x01(\tR\x02id:\x0f\x8a\xb5\x18\vbuildsignalB\x82\x01\n" +
+	"\x02id\x18\x01 \x01(\tR\x02id\x12\x1d\n" +
+	"\n" +
+	"queue_name\x18\x02 \x01(\tR\tqueueName:\x0f\x8a\xb5\x18\vbuildsignalB\x82\x01\n" +
 	"+com.uber.submitqueue.stovepipe.messagequeueB\x10BuildSignalProtoP\x01Z?github.com/uber/submitqueue/stovepipe/core/messagequeue/protopbb\x06proto3"
 
 var (
diff --git a/stovepipe/core/messagequeue/protopb/process.pb.go b/stovepipe/core/messagequeue/protopb/process.pb.go
index 5cffa3c2..943baa08 100644
--- a/stovepipe/core/messagequeue/protopb/process.pb.go
+++ b/stovepipe/core/messagequeue/protopb/process.pb.go
@@ -44,7 +44,11 @@ const (
 type ProcessRequest struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// id is the minted request id to process. Format: "request/<queue>/<counter>".
-	Id            string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
+	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
+	// queue_name is the name of the queue processing the request, carried so
+	// the consumer can route by queue without loading state first. Empty on
+	// payloads written before the field existed.
+	QueueName     string `protobuf:"bytes,2,opt,name=queue_name,json=queueName,proto3" json:"queue_name,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -86,13 +90,22 @@ func (x *ProcessRequest) GetId() string {
 	return ""
 }
 
+func (x *ProcessRequest) GetQueueName() string {
+	if x != nil {
+		return x.QueueName
+	}
+	return ""
+}
+
 var File_process_proto protoreflect.FileDescriptor
 
 const file_process_proto_rawDesc = "" +
 	"\n" +
-	"\rprocess.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"-\n" +
+	"\rprocess.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"L\n" +
 	"\x0eProcessRequest\x12\x0e\n" +
-	"\x02id\x18\x01 \x01(\tR\x02id:\v\x8a\xb5\x18\aprocessB~\n" +
+	"\x02id\x18\x01 \x01(\tR\x02id\x12\x1d\n" +
+	"\n" +
+	"queue_name\x18\x02 \x01(\tR\tqueueName:\v\x8a\xb5\x18\aprocessB~\n" +
 	"+com.uber.submitqueue.stovepipe.messagequeueB\fProcessProtoP\x01Z?github.com/uber/submitqueue/stovepipe/core/messagequeue/protopbb\x06proto3"
 
 var (
diff --git a/stovepipe/core/messagequeue/protopb/record.pb.go b/stovepipe/core/messagequeue/protopb/record.pb.go
index 5c85e7c8..0d99d565 100644
--- a/stovepipe/core/messagequeue/protopb/record.pb.go
+++ b/stovepipe/core/messagequeue/protopb/record.pb.go
@@ -47,7 +47,11 @@ type Record struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// id is the request id whose build reached a terminal status. Format:
 	// "request/<queue>/<counter>" (entity.Request.ID).
-	Id            string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
+	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
+	// queue_name is the name of the queue processing the request, carried so
+	// the consumer can route by queue without loading state first. Empty on
+	// payloads written before the field existed.
+	QueueName     string `protobuf:"bytes,2,opt,name=queue_name,json=queueName,proto3" json:"queue_name,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -89,13 +93,22 @@ func (x *Record) GetId() string {
 	return ""
 }
 
+func (x *Record) GetQueueName() string {
+	if x != nil {
+		return x.QueueName
+	}
+	return ""
+}
+
 var File_record_proto protoreflect.FileDescriptor
 
 const file_record_proto_rawDesc = "" +
 	"\n" +
-	"\frecord.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"$\n" +
+	"\frecord.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"C\n" +
 	"\x06Record\x12\x0e\n" +
-	"\x02id\x18\x01 \x01(\tR\x02id:\n" +
+	"\x02id\x18\x01 \x01(\tR\x02id\x12\x1d\n" +
+	"\n" +
+	"queue_name\x18\x02 \x01(\tR\tqueueName:\n" +
 	"\x8a\xb5\x18\x06recordB}\n" +
 	"+com.uber.submitqueue.stovepipe.messagequeueB\vRecordProtoP\x01Z?github.com/uber/submitqueue/stovepipe/core/messagequeue/protopbb\x06proto3"
 
diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md
index 3a768b7f..530b0787 100644
--- a/stovepipe/extension/storage/README.md
+++ b/stovepipe/extension/storage/README.md
@@ -1,6 +1,10 @@
 # Storage
 
-Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`). Implementations live under `extension/storage/<impl>/`. This is a separate contract from `submitqueue/extension/storage` — same shape and conventions by design, but its own interfaces and its own `ErrNotFound`/`ErrAlreadyExists`/`ErrVersionMismatch` sentinels, since Stovepipe and SubmitQueue are independent domains. `ErrVersionMismatch` is declared as a retryable infrastructure error so callers can return it without reclassifying it.
+Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`). Implementations live under `extension/storage/<impl>/`.
+
+The aggregate is resolved per queue through a factory keyed by queue name, mirroring the extension contract and the [submitqueue storage seam](../../../submitqueue/extension/storage/README.md#queue-scoped-resolution): a resolved instance is bound to its queue, entity arguments whose queue disagrees with the binding are rejected, and reads never surface another queue's records. Stovepipe has no cross-queue read paths, so every store — including `QueueStore`, whose row key is the queue name itself — lives inside the queue-scoped aggregate; there is no global remainder.
+
+This is a separate contract from `submitqueue/extension/storage` — same shape and conventions by design, but its own interfaces and its own `ErrNotFound`/`ErrAlreadyExists`/`ErrVersionMismatch` sentinels, since Stovepipe and SubmitQueue are independent domains. `ErrVersionMismatch` is declared as a retryable infrastructure error so callers can return it without reclassifying it.
 
 ## Optimistic locking contract
 
diff --git a/stovepipe/extension/storage/mock/request_uri_store_mock.go b/stovepipe/extension/storage/mock/request_uri_store_mock.go
index e50bf237..203a016c 100644
--- a/stovepipe/extension/storage/mock/request_uri_store_mock.go
+++ b/stovepipe/extension/storage/mock/request_uri_store_mock.go
@@ -41,30 +41,30 @@ func (m *MockRequestURIStore) EXPECT() *MockRequestURIStoreMockRecorder {
 }
 
 // Create mocks base method.
-func (m *MockRequestURIStore) Create(ctx context.Context, queue, uri, id string) error {
+func (m *MockRequestURIStore) Create(ctx context.Context, uri, id string) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "Create", ctx, queue, uri, id)
+	ret := m.ctrl.Call(m, "Create", ctx, uri, id)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // Create indicates an expected call of Create.
-func (mr *MockRequestURIStoreMockRecorder) Create(ctx, queue, uri, id any) *gomock.Call {
+func (mr *MockRequestURIStoreMockRecorder) Create(ctx, uri, id any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockRequestURIStore)(nil).Create), ctx, queue, uri, id)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockRequestURIStore)(nil).Create), ctx, uri, id)
 }
 
 // GetIDByURI mocks base method.
-func (m *MockRequestURIStore) GetIDByURI(ctx context.Context, queue, uri string) (string, error) {
+func (m *MockRequestURIStore) GetIDByURI(ctx context.Context, uri string) (string, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetIDByURI", ctx, queue, uri)
+	ret := m.ctrl.Call(m, "GetIDByURI", ctx, uri)
 	ret0, _ := ret[0].(string)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // GetIDByURI indicates an expected call of GetIDByURI.
-func (mr *MockRequestURIStoreMockRecorder) GetIDByURI(ctx, queue, uri any) *gomock.Call {
+func (mr *MockRequestURIStoreMockRecorder) GetIDByURI(ctx, uri any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetIDByURI", reflect.TypeOf((*MockRequestURIStore)(nil).GetIDByURI), ctx, queue, uri)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetIDByURI", reflect.TypeOf((*MockRequestURIStore)(nil).GetIDByURI), ctx, uri)
 }
diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go
index b2b8f013..f2f308d0 100644
--- a/stovepipe/extension/storage/mock/storage_mock.go
+++ b/stovepipe/extension/storage/mock/storage_mock.go
@@ -16,6 +16,45 @@ import (
 	gomock "go.uber.org/mock/gomock"
 )
 
+// MockFactory is a mock of Factory interface.
+type MockFactory struct {
+	ctrl     *gomock.Controller
+	recorder *MockFactoryMockRecorder
+	isgomock struct{}
+}
+
+// MockFactoryMockRecorder is the mock recorder for MockFactory.
+type MockFactoryMockRecorder struct {
+	mock *MockFactory
+}
+
+// NewMockFactory creates a new mock instance.
+func NewMockFactory(ctrl *gomock.Controller) *MockFactory {
+	mock := &MockFactory{ctrl: ctrl}
+	mock.recorder = &MockFactoryMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockFactory) EXPECT() *MockFactoryMockRecorder {
+	return m.recorder
+}
+
+// For mocks base method.
+func (m *MockFactory) For(config storage.Config) (storage.Storage, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "For", config)
+	ret0, _ := ret[0].(storage.Storage)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// For indicates an expected call of For.
+func (mr *MockFactoryMockRecorder) For(config any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockFactory)(nil).For), config)
+}
+
 // MockStorage is a mock of Storage interface.
 type MockStorage struct {
 	ctrl     *gomock.Controller
@@ -40,20 +79,6 @@ func (m *MockStorage) EXPECT() *MockStorageMockRecorder {
 	return m.recorder
 }
 
-// Close mocks base method.
-func (m *MockStorage) Close() error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "Close")
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// Close indicates an expected call of Close.
-func (mr *MockStorageMockRecorder) Close() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockStorage)(nil).Close))
-}
-
 // GetBuildStore mocks base method.
 func (m *MockStorage) GetBuildStore() storage.BuildStore {
 	m.ctrl.T.Helper()
diff --git a/stovepipe/extension/storage/mysql/build_store.go b/stovepipe/extension/storage/mysql/build_store.go
index 57847813..ab079a04 100644
--- a/stovepipe/extension/storage/mysql/build_store.go
+++ b/stovepipe/extension/storage/mysql/build_store.go
@@ -30,11 +30,14 @@ import (
 type buildStore struct {
 	db    *sql.DB
 	scope tally.Scope
+	// queue is the queue name this store instance is bound to; every read and
+	// write is scoped to it.
+	queue string
 }
 
 // NewBuildStore creates a new MySQL-backed BuildStore.
-func NewBuildStore(db *sql.DB, scope tally.Scope) storage.BuildStore {
-	return &buildStore{db: db, scope: scope}
+func NewBuildStore(db *sql.DB, scope tally.Scope, queue string) storage.BuildStore {
+	return &buildStore{db: db, scope: scope, queue: queue}
 }
 
 // Create persists a new build. Returns ErrAlreadyExists if the build ID already exists.
@@ -43,8 +46,9 @@ func (b *buildStore) Create(ctx context.Context, build entity.Build) (retErr err
 	defer func() { op.Complete(retErr) }()
 
 	_, err := b.db.ExecContext(ctx,
-		`INSERT INTO build (id, request_id, status, version)
-		 VALUES (?, ?, ?, ?)`,
+		`INSERT INTO build (queue, id, request_id, status, version)
+		 VALUES (?, ?, ?, ?, ?)`,
+		b.queue,
 		build.ID,
 		build.RequestID,
 		build.Status,
@@ -68,8 +72,8 @@ func (b *buildStore) Get(ctx context.Context, id string) (ret entity.Build, retE
 	var build entity.Build
 	err := b.db.QueryRowContext(ctx,
 		`SELECT id, request_id, status, version
-		 FROM build WHERE id = ?`,
-		id,
+		 FROM build WHERE queue = ? AND id = ?`,
+		b.queue, id,
 	).Scan(
 		&build.ID,
 		&build.RequestID,
@@ -98,9 +102,10 @@ func (b *buildStore) Update(ctx context.Context, build entity.Build, oldVersion,
 	result, err := b.db.ExecContext(ctx,
 		`UPDATE build
 		 SET status = ?, version = ?
-		 WHERE id = ? AND version = ?`,
+		 WHERE queue = ? AND id = ? AND version = ?`,
 		build.Status,
 		newVersion,
+		b.queue,
 		build.ID,
 		oldVersion,
 	)
diff --git a/stovepipe/extension/storage/mysql/build_store_test.go b/stovepipe/extension/storage/mysql/build_store_test.go
index f9425986..6e6e5911 100644
--- a/stovepipe/extension/storage/mysql/build_store_test.go
+++ b/stovepipe/extension/storage/mysql/build_store_test.go
@@ -35,7 +35,7 @@ func setupBuildStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, storage.BuildS
 	db, mock, err := sqlmock.New()
 	require.NoError(t, err)
 
-	store := NewBuildStore(db, testMetrics())
+	store := NewBuildStore(db, testMetrics(), "monorepo/main")
 
 	return db, mock, store
 }
@@ -58,7 +58,7 @@ func TestBuildStore_Create(t *testing.T) {
 			name: "success",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO build").
-					WithArgs(build.ID, build.RequestID, build.Status, build.Version).
+					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
@@ -66,7 +66,7 @@ func TestBuildStore_Create(t *testing.T) {
 			name: "duplicate id returns ErrAlreadyExists",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO build").
-					WithArgs(build.ID, build.RequestID, build.Status, build.Version).
+					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version).
 					WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry})
 			},
 			wantErr:   true,
@@ -76,7 +76,7 @@ func TestBuildStore_Create(t *testing.T) {
 			name: "other exec error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO build").
-					WithArgs(build.ID, build.RequestID, build.Status, build.Version).
+					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -127,7 +127,7 @@ func TestBuildStore_Get(t *testing.T) {
 				rows := sqlmock.NewRows([]string{"id", "request_id", "status", "version"}).
 					AddRow(want.ID, want.RequestID, string(want.Status), want.Version)
 				mock.ExpectQuery("SELECT id, request_id, status, version").
-					WithArgs(want.ID).
+					WithArgs("monorepo/main", want.ID).
 					WillReturnRows(rows)
 			},
 			want: want,
@@ -137,7 +137,7 @@ func TestBuildStore_Get(t *testing.T) {
 			id:   "missing",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectQuery("SELECT id, request_id, status, version").
-					WithArgs("missing").
+					WithArgs("monorepo/main", "missing").
 					WillReturnError(sql.ErrNoRows)
 			},
 			wantErr:   true,
@@ -148,7 +148,7 @@ func TestBuildStore_Get(t *testing.T) {
 			id:   "bad",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectQuery("SELECT id, request_id, status, version").
-					WithArgs("bad").
+					WithArgs("monorepo/main", "bad").
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -191,7 +191,7 @@ func TestBuildStore_Update(t *testing.T) {
 			name: "success",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE build").
-					WithArgs(build.Status, newVersion, build.ID, oldVersion).
+					WithArgs(build.Status, newVersion, "monorepo/main", build.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
@@ -199,7 +199,7 @@ func TestBuildStore_Update(t *testing.T) {
 			name: "version mismatch",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE build").
-					WithArgs(build.Status, newVersion, build.ID, oldVersion).
+					WithArgs(build.Status, newVersion, "monorepo/main", build.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 0))
 			},
 			wantErr:   true,
@@ -209,7 +209,7 @@ func TestBuildStore_Update(t *testing.T) {
 			name: "exec error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE build").
-					WithArgs(build.Status, newVersion, build.ID, oldVersion).
+					WithArgs(build.Status, newVersion, "monorepo/main", build.ID, oldVersion).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -218,7 +218,7 @@ func TestBuildStore_Update(t *testing.T) {
 			name: "rows affected error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE build").
-					WithArgs(build.Status, newVersion, build.ID, oldVersion).
+					WithArgs(build.Status, newVersion, "monorepo/main", build.ID, oldVersion).
 					WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error")))
 			},
 			wantErr: true,
diff --git a/stovepipe/extension/storage/mysql/queue_store.go b/stovepipe/extension/storage/mysql/queue_store.go
index 6c050d52..f82a9e1e 100644
--- a/stovepipe/extension/storage/mysql/queue_store.go
+++ b/stovepipe/extension/storage/mysql/queue_store.go
@@ -29,11 +29,14 @@ import (
 type queueStore struct {
 	db    *sql.DB
 	scope tally.Scope
+	// queue is the queue name this store instance is bound to; every read and
+	// write is scoped to it.
+	queue string
 }
 
 // NewQueueStore creates a new MySQL-backed QueueStore.
-func NewQueueStore(db *sql.DB, scope tally.Scope) storage.QueueStore {
-	return &queueStore{db: db, scope: scope}
+func NewQueueStore(db *sql.DB, scope tally.Scope, queue string) storage.QueueStore {
+	return &queueStore{db: db, scope: scope, queue: queue}
 }
 
 // Create persists a new queue row. Returns ErrAlreadyExists if the name already exists.
@@ -41,6 +44,10 @@ func (q *queueStore) Create(ctx context.Context, queue entity.Queue) (retErr err
 	op := metrics.Begin(q.scope, "create", metrics.StorageLatencyBuckets)
 	defer func() { op.Complete(retErr) }()
 
+	if queue.Name != q.queue {
+		return fmt.Errorf("queue row name %q does not match the store's bound queue %q", queue.Name, q.queue)
+	}
+
 	_, err := q.db.ExecContext(ctx,
 		`INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_id, version)
 		 VALUES (?, ?, ?, ?, ?)`,
@@ -64,6 +71,10 @@ func (q *queueStore) Get(ctx context.Context, name string) (ret entity.Queue, re
 	op := metrics.Begin(q.scope, "get", metrics.StorageLatencyBuckets)
 	defer func() { op.Complete(retErr) }()
 
+	if name != q.queue {
+		return entity.Queue{}, fmt.Errorf("queue row name %q does not match the store's bound queue %q: %w", name, q.queue, storage.ErrNotFound)
+	}
+
 	var queue entity.Queue
 	err := q.db.QueryRowContext(ctx,
 		"SELECT name, last_green_uri, in_flight_count, latest_request_id, version FROM queue WHERE name = ?",
@@ -92,6 +103,10 @@ func (q *queueStore) Update(ctx context.Context, queue entity.Queue, oldVersion,
 	op := metrics.Begin(q.scope, "update", metrics.StorageLatencyBuckets)
 	defer func() { op.Complete(retErr) }()
 
+	if queue.Name != q.queue {
+		return fmt.Errorf("queue row name %q does not match the store's bound queue %q", queue.Name, q.queue)
+	}
+
 	result, err := q.db.ExecContext(ctx,
 		`UPDATE queue
 		 SET last_green_uri = ?, in_flight_count = ?, latest_request_id = ?, version = ?
diff --git a/stovepipe/extension/storage/mysql/queue_store_test.go b/stovepipe/extension/storage/mysql/queue_store_test.go
index b766d2ab..d5507ea8 100644
--- a/stovepipe/extension/storage/mysql/queue_store_test.go
+++ b/stovepipe/extension/storage/mysql/queue_store_test.go
@@ -35,7 +35,7 @@ func setupQueueStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, storage.QueueS
 	db, mock, err := sqlmock.New()
 	require.NoError(t, err)
 
-	store := NewQueueStore(db, testMetrics())
+	store := NewQueueStore(db, testMetrics(), "monorepo/main")
 
 	return db, mock, store
 }
@@ -136,10 +136,10 @@ func TestQueueStore_Get(t *testing.T) {
 		},
 		{
 			name:      "not found",
-			queueName: "missing",
+			queueName: want.Name,
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version FROM queue").
-					WithArgs("missing").
+					WithArgs(want.Name).
 					WillReturnError(sql.ErrNoRows)
 			},
 			wantErr:   true,
@@ -147,10 +147,10 @@ func TestQueueStore_Get(t *testing.T) {
 		},
 		{
 			name:      "query error",
-			queueName: "bad",
+			queueName: want.Name,
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version FROM queue").
-					WithArgs("bad").
+					WithArgs(want.Name).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
diff --git a/stovepipe/extension/storage/mysql/request_store.go b/stovepipe/extension/storage/mysql/request_store.go
index bfc1627b..0c2492b1 100644
--- a/stovepipe/extension/storage/mysql/request_store.go
+++ b/stovepipe/extension/storage/mysql/request_store.go
@@ -35,11 +35,14 @@ const mysqlErrDuplicateEntry = 1062
 type requestStore struct {
 	db    *sql.DB
 	scope tally.Scope
+	// queue is the queue name this store instance is bound to; every read and
+	// write is scoped to it.
+	queue string
 }
 
 // NewRequestStore creates a new MySQL-backed RequestStore.
-func NewRequestStore(db *sql.DB, scope tally.Scope) storage.RequestStore {
-	return &requestStore{db: db, scope: scope}
+func NewRequestStore(db *sql.DB, scope tally.Scope, queue string) storage.RequestStore {
+	return &requestStore{db: db, scope: scope, queue: queue}
 }
 
 // Create persists a new request. Returns ErrAlreadyExists if the request ID already exists.
@@ -47,6 +50,10 @@ func (r *requestStore) Create(ctx context.Context, request entity.Request) (retE
 	op := metrics.Begin(r.scope, "create", metrics.StorageLatencyBuckets)
 	defer func() { op.Complete(retErr) }()
 
+	if request.Queue != r.queue {
+		return fmt.Errorf("request %s queue %q does not match the store's bound queue %q", request.ID, request.Queue, r.queue)
+	}
+
 	_, err := r.db.ExecContext(ctx,
 		`INSERT INTO request (id, queue, uri, state, build_strategy, base_uri, version)
 		 VALUES (?, ?, ?, ?, ?, ?, ?)`,
@@ -76,8 +83,8 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request,
 	var req entity.Request
 	err := r.db.QueryRowContext(ctx,
 		`SELECT id, queue, uri, state, build_strategy, base_uri, version
-		 FROM request WHERE id = ?`,
-		id,
+		 FROM request WHERE queue = ? AND id = ?`,
+		r.queue, id,
 	).Scan(
 		&req.ID,
 		&req.Queue,
@@ -106,15 +113,20 @@ func (r *requestStore) Update(ctx context.Context, request entity.Request, oldVe
 	op := metrics.Begin(r.scope, "update", metrics.StorageLatencyBuckets)
 	defer func() { op.Complete(retErr) }()
 
+	if request.Queue != r.queue {
+		return fmt.Errorf("request %s queue %q does not match the store's bound queue %q", request.ID, request.Queue, r.queue)
+	}
+
 	result, err := r.db.ExecContext(ctx,
 		`UPDATE request
 		 SET uri = ?, state = ?, build_strategy = ?, base_uri = ?, version = ?
-		 WHERE id = ? AND version = ?`,
+		 WHERE queue = ? AND id = ? AND version = ?`,
 		request.URI,
 		request.State,
 		request.BuildStrategy,
 		request.BaseURI,
 		newVersion,
+		request.Queue,
 		request.ID,
 		oldVersion,
 	)
diff --git a/stovepipe/extension/storage/mysql/request_store_test.go b/stovepipe/extension/storage/mysql/request_store_test.go
index b12465ce..0fc657cf 100644
--- a/stovepipe/extension/storage/mysql/request_store_test.go
+++ b/stovepipe/extension/storage/mysql/request_store_test.go
@@ -36,7 +36,7 @@ func setupRequestStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, storage.Requ
 	db, mock, err := sqlmock.New()
 	require.NoError(t, err)
 
-	store := NewRequestStore(db, testMetrics())
+	store := NewRequestStore(db, testMetrics(), "monorepo/main")
 
 	return db, mock, store
 }
@@ -134,7 +134,7 @@ func TestRequestStore_Get(t *testing.T) {
 				rows := sqlmock.NewRows([]string{"id", "queue", "uri", "state", "build_strategy", "base_uri", "version"}).
 					AddRow(want.ID, want.Queue, want.URI, string(want.State), string(want.BuildStrategy), want.BaseURI, want.Version)
 				mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, version").
-					WithArgs(want.ID).
+					WithArgs("monorepo/main", want.ID).
 					WillReturnRows(rows)
 			},
 			want: want,
@@ -144,7 +144,7 @@ func TestRequestStore_Get(t *testing.T) {
 			id:   "missing",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, version").
-					WithArgs("missing").
+					WithArgs("monorepo/main", "missing").
 					WillReturnError(sql.ErrNoRows)
 			},
 			wantErr:   true,
@@ -155,7 +155,7 @@ func TestRequestStore_Get(t *testing.T) {
 			id:   "bad",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, version").
-					WithArgs("bad").
+					WithArgs("monorepo/main", "bad").
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -187,6 +187,7 @@ func TestRequestStore_Get(t *testing.T) {
 func TestRequestStore_Update(t *testing.T) {
 	request := entity.Request{
 		ID:            "request/monorepo/main/1",
+		Queue:         "monorepo/main",
 		URI:           "git://remote/monorepo/main/deadbeef",
 		State:         entity.RequestStateProcessing,
 		BuildStrategy: entity.BuildStrategyFull,
@@ -204,7 +205,7 @@ func TestRequestStore_Update(t *testing.T) {
 			name: "success",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE request").
-					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, newVersion, request.ID, oldVersion).
+					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, newVersion, request.Queue, request.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
@@ -212,7 +213,7 @@ func TestRequestStore_Update(t *testing.T) {
 			name: "version mismatch",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE request").
-					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, newVersion, request.ID, oldVersion).
+					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, newVersion, request.Queue, request.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 0))
 			},
 			wantErr:   true,
@@ -222,7 +223,7 @@ func TestRequestStore_Update(t *testing.T) {
 			name: "exec error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE request").
-					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, newVersion, request.ID, oldVersion).
+					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, newVersion, request.Queue, request.ID, oldVersion).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -231,7 +232,7 @@ func TestRequestStore_Update(t *testing.T) {
 			name: "rows affected error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE request").
-					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, newVersion, request.ID, oldVersion).
+					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, newVersion, request.Queue, request.ID, oldVersion).
 					WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error")))
 			},
 			wantErr: true,
diff --git a/stovepipe/extension/storage/mysql/request_uri_store.go b/stovepipe/extension/storage/mysql/request_uri_store.go
index 8106596f..bd1c5318 100644
--- a/stovepipe/extension/storage/mysql/request_uri_store.go
+++ b/stovepipe/extension/storage/mysql/request_uri_store.go
@@ -34,19 +34,23 @@ const requestURIInitialVersion = 1
 type requestURIStore struct {
 	db    *sql.DB
 	scope tally.Scope
+	// queue is the queue name this store instance is bound to; every read and
+	// write is scoped to it.
+	queue string
 }
 
 // NewRequestURIStore creates a new MySQL-backed RequestURIStore.
-func NewRequestURIStore(db *sql.DB, scope tally.Scope) storage.RequestURIStore {
-	return &requestURIStore{db: db, scope: scope}
+func NewRequestURIStore(db *sql.DB, scope tally.Scope, queue string) storage.RequestURIStore {
+	return &requestURIStore{db: db, scope: scope, queue: queue}
 }
 
-// Create records the (queue, uri) -> id reverse index. Returns ErrAlreadyExists if (queue, uri)
-// is already mapped to a request.
-func (r *requestURIStore) Create(ctx context.Context, queue, uri, id string) (retErr error) {
+// Create records the bound queue's uri -> id reverse index. Returns
+// ErrAlreadyExists if the queue already maps the uri to a request.
+func (r *requestURIStore) Create(ctx context.Context, uri, id string) (retErr error) {
 	op := metrics.Begin(r.scope, "create", metrics.StorageLatencyBuckets)
 	defer func() { op.Complete(retErr) }()
 
+	queue := r.queue
 	_, err := r.db.ExecContext(ctx,
 		"INSERT INTO request_uri (queue, uri, request_id, version) VALUES (?, ?, ?, ?)",
 		queue, uri, id, requestURIInitialVersion,
@@ -61,11 +65,13 @@ func (r *requestURIStore) Create(ctx context.Context, queue, uri, id string) (re
 	return nil
 }
 
-// GetIDByURI returns the id of the request validating (queue, uri). Returns ErrNotFound if absent.
-func (r *requestURIStore) GetIDByURI(ctx context.Context, queue, uri string) (ret string, retErr error) {
+// GetIDByURI returns the id of the request validating the bound queue's uri.
+// Returns ErrNotFound if absent.
+func (r *requestURIStore) GetIDByURI(ctx context.Context, uri string) (ret string, retErr error) {
 	op := metrics.Begin(r.scope, "get_id_by_uri", metrics.StorageLatencyBuckets)
 	defer func() { op.Complete(retErr) }()
 
+	queue := r.queue
 	var id string
 	err := r.db.QueryRowContext(ctx,
 		"SELECT request_id FROM request_uri WHERE queue = ? AND uri = ?",
diff --git a/stovepipe/extension/storage/mysql/request_uri_store_test.go b/stovepipe/extension/storage/mysql/request_uri_store_test.go
index 1e7b3765..fae0185f 100644
--- a/stovepipe/extension/storage/mysql/request_uri_store_test.go
+++ b/stovepipe/extension/storage/mysql/request_uri_store_test.go
@@ -34,7 +34,7 @@ func setupRequestURIStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, storage.R
 	db, mock, err := sqlmock.New()
 	require.NoError(t, err)
 
-	store := NewRequestURIStore(db, testMetrics())
+	store := NewRequestURIStore(db, testMetrics(), "monorepo/main")
 
 	return db, mock, store
 }
@@ -84,7 +84,7 @@ func TestRequestURIStore_Create(t *testing.T) {
 
 			tt.setup(mock)
 
-			err := store.Create(context.Background(), queue, uri, id)
+			err := store.Create(context.Background(), uri, id)
 			if tt.wantErr {
 				require.Error(t, err)
 				if tt.wantErrIs != nil {
@@ -154,7 +154,7 @@ func TestRequestURIStore_GetIDByURI(t *testing.T) {
 
 			tt.setup(mock)
 
-			got, err := store.GetIDByURI(context.Background(), tt.queue, tt.uri)
+			got, err := store.GetIDByURI(context.Background(), tt.uri)
 			if tt.wantErr {
 				require.Error(t, err)
 				if tt.wantErrIs != nil {
diff --git a/stovepipe/extension/storage/mysql/schema/build.sql b/stovepipe/extension/storage/mysql/schema/build.sql
index 21106ded..4c60f17c 100644
--- a/stovepipe/extension/storage/mysql/schema/build.sql
+++ b/stovepipe/extension/storage/mysql/schema/build.sql
@@ -1,9 +1,10 @@
 -- build holds one CI build triggered for a request's commit. id is the runner-assigned build id
 -- minted at Trigger (e.g. a Buildkite build number), opaque and never parsed or derived.
 CREATE TABLE IF NOT EXISTS build (
+    queue       VARCHAR(255) NOT NULL,
     id          VARCHAR(255) NOT NULL,
     request_id  VARCHAR(255) NOT NULL,
     status      VARCHAR(64)  NOT NULL,
     version     INT          NOT NULL,
-    PRIMARY KEY (id)
+    PRIMARY KEY (queue, id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
diff --git a/stovepipe/extension/storage/mysql/schema/request.sql b/stovepipe/extension/storage/mysql/schema/request.sql
index 863288ae..14c76687 100644
--- a/stovepipe/extension/storage/mysql/schema/request.sql
+++ b/stovepipe/extension/storage/mysql/schema/request.sql
@@ -2,12 +2,12 @@
 -- VCS-agnostic commit locator; it may be empty until SourceControl resolution is wired in.
 -- No timestamps: created/updated times are not part of the Request entity.
 CREATE TABLE IF NOT EXISTS request (
-    id              VARCHAR(255) NOT NULL,
     queue           VARCHAR(255) NOT NULL,
+    id              VARCHAR(255) NOT NULL,
     uri             VARCHAR(255) NOT NULL,
     state           VARCHAR(64)  NOT NULL,
     build_strategy  VARCHAR(64)  NOT NULL DEFAULT '',
     base_uri        VARCHAR(255) NOT NULL DEFAULT '',
     version         INT          NOT NULL,
-    PRIMARY KEY (id)
+    PRIMARY KEY (queue, id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
diff --git a/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go
index eac6fdcc..f670a025 100644
--- a/stovepipe/extension/storage/mysql/storage.go
+++ b/stovepipe/extension/storage/mysql/storage.go
@@ -16,30 +16,57 @@ package mysql
 
 import (
 	"database/sql"
+	"fmt"
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/uber-go/tally"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 )
 
+// Storage is the MySQL storage backend. It owns the shared connection pool and
+// binds queue-scoped store aggregates over the shared tables on demand via For.
+// The wiring layer adapts For into the storage.Factory seam; per-queue backend
+// routing stays a host decision.
+type Storage struct {
+	db    *sql.DB
+	scope tally.Scope
+}
+
+// NewStorage creates a new MySQL storage backend over the given connection pool.
+func NewStorage(db *sql.DB, scope tally.Scope) (*Storage, error) {
+	return &Storage{db: db, scope: scope}, nil
+}
+
+// For returns the queue-scoped store aggregate bound to queueName over the
+// shared pool. Every store the aggregate hands back reads and writes only that
+// queue's records.
+func (s *Storage) For(queueName string) (storage.Storage, error) {
+	if queueName == "" {
+		return nil, fmt.Errorf("queue name must not be empty")
+	}
+	return &mysqlStorage{
+		requestStore:    NewRequestStore(s.db, s.scope.SubScope("request_store"), queueName),
+		requestURIStore: NewRequestURIStore(s.db, s.scope.SubScope("request_uri_store"), queueName),
+		queueStore:      NewQueueStore(s.db, s.scope.SubScope("queue_store"), queueName),
+		buildStore:      NewBuildStore(s.db, s.scope.SubScope("build_store"), queueName),
+	}, nil
+}
+
+// Close closes the underlying database connection.
+func (s *Storage) Close() error {
+	return s.db.Close()
+}
+
+// mysqlStorage is the queue-scoped store aggregate returned by For.
 type mysqlStorage struct {
-	db              *sql.DB
 	requestStore    storage.RequestStore
 	requestURIStore storage.RequestURIStore
 	queueStore      storage.QueueStore
 	buildStore      storage.BuildStore
 }
 
-// NewStorage creates a new MySQL-backed storage.
-func NewStorage(db *sql.DB, scope tally.Scope) (storage.Storage, error) {
-	return &mysqlStorage{
-		db:              db,
-		requestStore:    NewRequestStore(db, scope.SubScope("request_store")),
-		requestURIStore: NewRequestURIStore(db, scope.SubScope("request_uri_store")),
-		queueStore:      NewQueueStore(db, scope.SubScope("queue_store")),
-		buildStore:      NewBuildStore(db, scope.SubScope("build_store")),
-	}, nil
-}
+// Verify mysqlStorage implements the queue-scoped aggregate at compile time.
+var _ storage.Storage = (*mysqlStorage)(nil)
 
 // GetRequestStore returns the MySQL-backed RequestStore.
 func (f *mysqlStorage) GetRequestStore() storage.RequestStore {
@@ -60,8 +87,3 @@ func (f *mysqlStorage) GetQueueStore() storage.QueueStore {
 func (f *mysqlStorage) GetBuildStore() storage.BuildStore {
 	return f.buildStore
 }
-
-// Close closes the underlying database connection.
-func (f *mysqlStorage) Close() error {
-	return f.db.Close()
-}
diff --git a/stovepipe/extension/storage/mysql/storage_test.go b/stovepipe/extension/storage/mysql/storage_test.go
index 7cd7963d..04405ff9 100644
--- a/stovepipe/extension/storage/mysql/storage_test.go
+++ b/stovepipe/extension/storage/mysql/storage_test.go
@@ -36,10 +36,15 @@ func TestNewStorage(t *testing.T) {
 	s, err := NewStorage(db, testMetrics())
 	require.NoError(t, err)
 
-	assert.NotNil(t, s.GetRequestStore())
-	assert.NotNil(t, s.GetRequestURIStore())
-	assert.NotNil(t, s.GetQueueStore())
-	assert.NotNil(t, s.GetBuildStore())
+	bound, err := s.For("monorepo/main")
+	require.NoError(t, err)
+	assert.NotNil(t, bound.GetRequestStore())
+	assert.NotNil(t, bound.GetRequestURIStore())
+	assert.NotNil(t, bound.GetQueueStore())
+	assert.NotNil(t, bound.GetBuildStore())
+
+	_, err = s.For("")
+	assert.Error(t, err, "resolving an empty queue name must fail")
 }
 
 func TestMysqlStorage_Close(t *testing.T) {
diff --git a/stovepipe/extension/storage/request_uri_store.go b/stovepipe/extension/storage/request_uri_store.go
index e5f9f6b7..f3dbf565 100644
--- a/stovepipe/extension/storage/request_uri_store.go
+++ b/stovepipe/extension/storage/request_uri_store.go
@@ -25,12 +25,13 @@ import (
 // table — the two are written independently (no cross-table transaction), keeping the contract
 // satisfiable by key/value backends. The caller orchestrates "create request, then map URI".
 type RequestURIStore interface {
-	// Create records that request id validates the commit uri for queue. Returns ErrAlreadyExists
-	// if (queue, uri) is already mapped — the signal a caller uses to detect that the commit is
-	// already being validated by another request.
-	Create(ctx context.Context, queue, uri, id string) error
+	// Create records that request id validates the commit uri for the bound
+	// queue. Returns ErrAlreadyExists if the queue already maps the uri — the
+	// signal a caller uses to detect that the commit is already being validated
+	// by another request.
+	Create(ctx context.Context, uri, id string) error
 
-	// GetIDByURI returns the id of the request validating (queue, uri).
+	// GetIDByURI returns the id of the request validating the bound queue's uri.
 	// Returns ErrNotFound if no request is mapped to that commit.
-	GetIDByURI(ctx context.Context, queue, uri string) (string, error)
+	GetIDByURI(ctx context.Context, uri string) (string, error)
 }
diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go
index 1d06ef73..6fa1e487 100644
--- a/stovepipe/extension/storage/storage.go
+++ b/stovepipe/extension/storage/storage.go
@@ -44,7 +44,30 @@ var ErrAlreadyExists = errors.New("record already exists")
 // retry or converge instead of overwriting a concurrent change. It is intrinsically a retryable infrastructure error.
 var ErrVersionMismatch = errs.NewRetryableError(errors.New("version mismatch"))
 
-// Storage is a factory interface that aggregates all entity stores into a single injectable dependency.
+// Config identifies the queue a Storage instance is resolved for. Like every
+// other extension config, it carries only the queue name — everything an
+// implementation needs beyond that is injected at construction by the
+// integrator.
+type Config struct {
+	// QueueName is the name of the queue whose data the resolved Storage is
+	// scoped to.
+	QueueName string
+}
+
+// Factory resolves the queue-scoped Storage aggregate for a queue. Mirrors the
+// extension contract: the host wiring decides which backend serves which
+// queue; implementations bind the queue over their backend so a resolved
+// instance can only read and write that queue's data. Stovepipe has no
+// cross-queue read paths, so every store lives inside the aggregate.
+type Factory interface {
+	// For returns the Storage aggregate bound to the queue named in config.
+	For(config Config) (Storage, error)
+}
+
+// Storage aggregates the queue-scoped entity stores into a single injectable
+// dependency. An instance is resolved per queue through Factory and is bound
+// to that queue: entity arguments whose queue disagrees with the binding are
+// rejected, and reads never surface another queue's records.
 type Storage interface {
 	// GetRequestStore returns the RequestStore instance.
 	GetRequestStore() RequestStore
@@ -57,7 +80,4 @@ type Storage interface {
 
 	// GetBuildStore returns the BuildStore instance.
 	GetBuildStore() BuildStore
-
-	// Close closes the storage and all underlying connections. Should only be called once at the end of the program.
-	Close() error
 }
diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go
index ee55fe5a..5ebc21b5 100644
--- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go
+++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go
@@ -50,15 +50,19 @@ func TestMySQLStorage(t *testing.T) {
 	schemaDir := testutil.SchemaDir("stovepipe/extension/storage/mysql/schema")
 	testutil.ApplySchema(t, log, db, schemaDir)
 
-	store, err := mysqlstorage.NewStorage(db, tally.NoopScope)
+	backend, err := mysqlstorage.NewStorage(db, tally.NoopScope)
 	require.NoError(t, err, "failed to create storage")
+	factory := mysqlFactory{backend: backend}
 
 	t.Run("RequestStore", func(t *testing.T) {
 		resetStorage(t, db)
+		bound, err := backend.For("monorepo/main")
+		require.NoError(t, err)
 		suite.Run(t, &MySQLRequestStoreSuite{
 			ctx:      ctx,
-			store:    store.GetRequestStore(),
-			uriStore: store.GetRequestURIStore(),
+			backend:  backend,
+			store:    bound.GetRequestStore(),
+			uriStore: bound.GetRequestURIStore(),
 		})
 	})
 
@@ -66,7 +70,7 @@ func TestMySQLStorage(t *testing.T) {
 		resetStorage(t, db)
 		testSuite := new(MySQLQueueStoreSuite)
 		testSuite.SetContext(ctx)
-		testSuite.SetQueueStore(store.GetQueueStore())
+		testSuite.SetFactory(factory)
 		testSuite.SetLogger(testutil.NewTestLogger(t))
 		suite.Run(t, testSuite)
 	})
@@ -75,7 +79,9 @@ func TestMySQLStorage(t *testing.T) {
 		resetStorage(t, db)
 		testSuite := new(MySQLBuildStoreSuite)
 		testSuite.SetContext(ctx)
-		testSuite.SetBuildStore(store.GetBuildStore())
+		bound, err := backend.For("contract/queue")
+		require.NoError(t, err)
+		testSuite.SetBuildStore(bound.GetBuildStore())
 		testSuite.SetLogger(testutil.NewTestLogger(t))
 		suite.Run(t, testSuite)
 	})
@@ -99,6 +105,7 @@ func resetStorage(t *testing.T, db *sql.DB) {
 type MySQLRequestStoreSuite struct {
 	suite.Suite
 	ctx      context.Context
+	backend  *mysqlstorage.Storage
 	store    storage.RequestStore
 	uriStore storage.RequestURIStore
 }
@@ -169,7 +176,7 @@ func (s *MySQLRequestStoreSuite) TestUpdateCAS() {
 }
 
 func (s *MySQLRequestStoreSuite) TestUpdateNotFoundIsVersionMismatch() {
-	missing := entity.Request{ID: "request/monorepo/main/missing", State: entity.RequestStateAccepted}
+	missing := entity.Request{ID: "request/monorepo/main/missing", Queue: "monorepo/main", State: entity.RequestStateAccepted}
 	err := s.store.Update(s.ctx, missing, 1, 2)
 	require.ErrorIs(s.T(), err, storage.ErrVersionMismatch)
 }
@@ -193,15 +200,15 @@ func (s *MySQLRequestStoreSuite) TestURIMappingCreateAndGet() {
 		uri   = "git://remote/monorepo/main/bbbb2222"
 		id    = "request/monorepo/main/3"
 	)
-	require.NoError(s.T(), s.uriStore.Create(s.ctx, queue, uri, id))
+	require.NoError(s.T(), s.uriStore.Create(s.ctx, uri, id))
 
-	got, err := s.uriStore.GetIDByURI(s.ctx, queue, uri)
+	got, err := s.uriStore.GetIDByURI(s.ctx, uri)
 	require.NoError(s.T(), err)
 	require.Equal(s.T(), id, got)
 }
 
 func (s *MySQLRequestStoreSuite) TestGetIDByURINotFound() {
-	_, err := s.uriStore.GetIDByURI(s.ctx, "monorepo/main", "git://remote/monorepo/main/unmapped")
+	_, err := s.uriStore.GetIDByURI(s.ctx, "git://remote/monorepo/main/unmapped")
 	require.True(s.T(), storage.IsNotFound(err))
 }
 
@@ -210,25 +217,33 @@ func (s *MySQLRequestStoreSuite) TestURIMappingDuplicate() {
 		queue = "monorepo/main"
 		uri   = "git://remote/monorepo/main/cccc3333"
 	)
-	require.NoError(s.T(), s.uriStore.Create(s.ctx, queue, uri, "request/monorepo/main/4"))
+	require.NoError(s.T(), s.uriStore.Create(s.ctx, uri, "request/monorepo/main/4"))
 
 	// A second request claiming the same (queue, uri) is rejected — the dedup signal.
-	err := s.uriStore.Create(s.ctx, queue, uri, "request/monorepo/main/5")
+	err := s.uriStore.Create(s.ctx, uri, "request/monorepo/main/5")
 	require.ErrorIs(s.T(), err, storage.ErrAlreadyExists)
 }
 
 func (s *MySQLRequestStoreSuite) TestURIMappingDistinctAcrossQueues() {
 	const uri = "git://remote/monorepo/shared/dddd4444"
-	require.NoError(s.T(), s.uriStore.Create(s.ctx, "queue-a", uri, "request/queue-a/1"))
-	require.NoError(s.T(), s.uriStore.Create(s.ctx, "queue-b", uri, "request/queue-b/1"))
+	boundA, err := s.backend.For("queue-a")
+	require.NoError(s.T(), err)
+	boundB, err := s.backend.For("queue-b")
+	require.NoError(s.T(), err)
+	require.NoError(s.T(), boundA.GetRequestURIStore().Create(s.ctx, uri, "request/queue-a/1"))
+	require.NoError(s.T(), boundB.GetRequestURIStore().Create(s.ctx, uri, "request/queue-b/1"))
 
-	idA, err := s.uriStore.GetIDByURI(s.ctx, "queue-a", uri)
+	idA, err := boundA.GetRequestURIStore().GetIDByURI(s.ctx, uri)
 	require.NoError(s.T(), err)
 	require.Equal(s.T(), "request/queue-a/1", idA)
 
-	idB, err := s.uriStore.GetIDByURI(s.ctx, "queue-b", uri)
+	idB, err := boundB.GetRequestURIStore().GetIDByURI(s.ctx, uri)
 	require.NoError(s.T(), err)
 	require.Equal(s.T(), "request/queue-b/1", idB)
+
+	// The other queue's mapping is invisible through this queue's binding.
+	_, err = boundA.GetRequestURIStore().GetIDByURI(s.ctx, "git://remote/monorepo/shared/only-b")
+	require.True(s.T(), storage.IsNotFound(err))
 }
 
 // MySQLQueueStoreSuite exercises the MySQL-backed QueueStore by embedding the shared contract suite.
@@ -240,3 +255,14 @@ type MySQLQueueStoreSuite struct {
 type MySQLBuildStoreSuite struct {
 	storagesuite.BuildStoreContractSuite
 }
+
+// mysqlFactory adapts the MySQL storage backend's queue binding to the
+// storage.Factory seam for the contract suite, mirroring the host wiring.
+type mysqlFactory struct {
+	backend *mysqlstorage.Storage
+}
+
+// For returns the queue-scoped store aggregate bound to the queue named in config.
+func (f mysqlFactory) For(config storage.Config) (storage.Storage, error) {
+	return f.backend.For(config.QueueName)
+}
diff --git a/test/integration/stovepipe/extension/storage/suite.go b/test/integration/stovepipe/extension/storage/suite.go
index 2af673eb..7d636db9 100644
--- a/test/integration/stovepipe/extension/storage/suite.go
+++ b/test/integration/stovepipe/extension/storage/suite.go
@@ -26,12 +26,14 @@ import (
 )
 
 // QueueStoreContractSuite defines contract tests for storage.QueueStore.
-// All QueueStore implementations must pass these tests.
+// All QueueStore implementations must pass these tests. Queue rows are keyed
+// by the queue name itself, so each test resolves the store bound to its own
+// test queue through the factory.
 type QueueStoreContractSuite struct {
 	suite.Suite
-	ctx        context.Context
-	queueStore storage.QueueStore
-	log        *testutil.TestLogger
+	ctx     context.Context
+	factory storage.Factory
+	log     *testutil.TestLogger
 }
 
 // SetContext sets the context for tests.
@@ -39,9 +41,16 @@ func (s *QueueStoreContractSuite) SetContext(ctx context.Context) {
 	s.ctx = ctx
 }
 
-// SetQueueStore provides the concrete QueueStore under test.
-func (s *QueueStoreContractSuite) SetQueueStore(store storage.QueueStore) {
-	s.queueStore = store
+// SetFactory provides the storage factory under test.
+func (s *QueueStoreContractSuite) SetFactory(factory storage.Factory) {
+	s.factory = factory
+}
+
+// storeFor resolves the QueueStore bound to the named queue.
+func (s *QueueStoreContractSuite) storeFor(name string) storage.QueueStore {
+	store, err := s.factory.For(storage.Config{QueueName: name})
+	s.Require().NoError(err)
+	return store.GetQueueStore()
 }
 
 // SetLogger sets the logger for tests.
@@ -58,12 +67,12 @@ func (s *QueueStoreContractSuite) TestQueueStore_Create() {
 	t := s.T()
 	const name = "contract/create"
 
-	require.NoError(t, s.queueStore.Create(s.ctx, entity.Queue{
+	require.NoError(t, s.storeFor(name).Create(s.ctx, entity.Queue{
 		Name:    name,
 		Version: 1,
 	}))
 
-	got, err := s.queueStore.Get(s.ctx, name)
+	got, err := s.storeFor(name).Get(s.ctx, name)
 	require.NoError(t, err)
 	assert.Equal(t, entity.Queue{
 		Name:            name,
@@ -83,9 +92,9 @@ func (s *QueueStoreContractSuite) TestQueueStore_CreateWithFields() {
 		LatestRequestID: "request/contract/defaults/99",
 		Version:         1,
 	}
-	require.NoError(t, s.queueStore.Create(s.ctx, toCreate))
+	require.NoError(t, s.storeFor(name).Create(s.ctx, toCreate))
 
-	got, err := s.queueStore.Get(s.ctx, name)
+	got, err := s.storeFor(name).Get(s.ctx, name)
 	require.NoError(t, err)
 	assert.Equal(t, toCreate, got)
 }
@@ -96,9 +105,9 @@ func (s *QueueStoreContractSuite) TestQueueStore_CreateAlreadyExists() {
 	const name = "contract/already-exists"
 
 	first := entity.Queue{Name: name, LatestRequestID: "request/contract/already-exists/3", Version: 1}
-	require.NoError(t, s.queueStore.Create(s.ctx, first))
+	require.NoError(t, s.storeFor(name).Create(s.ctx, first))
 
-	err := s.queueStore.Create(s.ctx, entity.Queue{
+	err := s.storeFor(name).Create(s.ctx, entity.Queue{
 		Name:            name,
 		LastGreenURI:    "git://remote/monorepo/main/ignored-on-race",
 		LatestRequestID: "request/contract/already-exists/500",
@@ -106,7 +115,7 @@ func (s *QueueStoreContractSuite) TestQueueStore_CreateAlreadyExists() {
 	})
 	assert.ErrorIs(t, err, storage.ErrAlreadyExists)
 
-	got, err := s.queueStore.Get(s.ctx, name)
+	got, err := s.storeFor(name).Get(s.ctx, name)
 	require.NoError(t, err)
 	assert.Equal(t, first, got)
 }
@@ -115,7 +124,7 @@ func (s *QueueStoreContractSuite) TestQueueStore_CreateAlreadyExists() {
 func (s *QueueStoreContractSuite) TestQueueStore_GetNotFound() {
 	t := s.T()
 
-	_, err := s.queueStore.Get(s.ctx, "contract/does-not-exist")
+	_, err := s.storeFor("contract/does-not-exist").Get(s.ctx, "contract/does-not-exist")
 	assert.True(t, storage.IsNotFound(err))
 }
 
@@ -125,22 +134,22 @@ func (s *QueueStoreContractSuite) TestQueueStore_UpdateCAS() {
 	const name = "contract/update-cas"
 
 	created := entity.Queue{Name: name, Version: 1}
-	require.NoError(t, s.queueStore.Create(s.ctx, created))
+	require.NoError(t, s.storeFor(name).Create(s.ctx, created))
 
 	updated := created
 	updated.LastGreenURI = "git://remote/monorepo/main/green-cccc"
 	updated.LatestRequestID = "request/contract/update-cas/42"
 	updated.InFlightCount = 1
-	require.NoError(t, s.queueStore.Update(s.ctx, updated, 1, 2))
+	require.NoError(t, s.storeFor(name).Update(s.ctx, updated, 1, 2))
 
-	got, err := s.queueStore.Get(s.ctx, name)
+	got, err := s.storeFor(name).Get(s.ctx, name)
 	require.NoError(t, err)
 	assert.Equal(t, updated.LastGreenURI, got.LastGreenURI)
 	assert.Equal(t, "request/contract/update-cas/42", got.LatestRequestID)
 	assert.Equal(t, int32(1), got.InFlightCount)
 	assert.Equal(t, int32(2), got.Version)
 
-	err = s.queueStore.Update(s.ctx, updated, 1, 2)
+	err = s.storeFor(name).Update(s.ctx, updated, 1, 2)
 	assert.ErrorIs(t, err, storage.ErrVersionMismatch)
 }
 
@@ -148,7 +157,7 @@ func (s *QueueStoreContractSuite) TestQueueStore_UpdateCAS() {
 func (s *QueueStoreContractSuite) TestQueueStore_UpdateNotFoundIsVersionMismatch() {
 	t := s.T()
 
-	err := s.queueStore.Update(s.ctx, entity.Queue{Name: "contract/missing"}, 1, 2)
+	err := s.storeFor("contract/missing").Update(s.ctx, entity.Queue{Name: "contract/missing"}, 1, 2)
 	assert.ErrorIs(t, err, storage.ErrVersionMismatch)
 }
 
@@ -157,15 +166,15 @@ func (s *QueueStoreContractSuite) TestQueueStore_UpdateSequentialCAS() {
 	t := s.T()
 	const name = "contract/sequential-cas"
 
-	require.NoError(t, s.queueStore.Create(s.ctx, entity.Queue{Name: name, Version: 1}))
+	require.NoError(t, s.storeFor(name).Create(s.ctx, entity.Queue{Name: name, Version: 1}))
 
 	v2 := entity.Queue{Name: name, LatestRequestID: "request/contract/sequential-cas/10", Version: 1}
-	require.NoError(t, s.queueStore.Update(s.ctx, v2, 1, 2))
+	require.NoError(t, s.storeFor(name).Update(s.ctx, v2, 1, 2))
 
 	v3 := entity.Queue{Name: name, LatestRequestID: "request/contract/sequential-cas/10", InFlightCount: 1, Version: 2}
-	require.NoError(t, s.queueStore.Update(s.ctx, v3, 2, 3))
+	require.NoError(t, s.storeFor(name).Update(s.ctx, v3, 2, 3))
 
-	got, err := s.queueStore.Get(s.ctx, name)
+	got, err := s.storeFor(name).Get(s.ctx, name)
 	require.NoError(t, err)
 	assert.Equal(t, "request/contract/sequential-cas/10", got.LatestRequestID)
 	assert.Equal(t, int32(1), got.InFlightCount)
@@ -360,10 +369,12 @@ func (s *QueueStoreContractSuite) TestQueueStore_QueueIsolation() {
 		nameB = "contract/isolation-b"
 	)
 
-	require.NoError(t, s.queueStore.Create(s.ctx, entity.Queue{Name: nameA, Version: 1}))
-	require.NoError(t, s.queueStore.Create(s.ctx, entity.Queue{Name: nameB, Version: 1}))
+	storeA := s.storeFor(nameA)
+	storeB := s.storeFor(nameB)
+	require.NoError(t, storeA.Create(s.ctx, entity.Queue{Name: nameA, Version: 1}))
+	require.NoError(t, storeB.Create(s.ctx, entity.Queue{Name: nameB, Version: 1}))
 
-	baseline, err := s.queueStore.Get(s.ctx, nameB)
+	baseline, err := storeB.Get(s.ctx, nameB)
 	require.NoError(t, err)
 
 	updatedA := entity.Queue{
@@ -373,9 +384,14 @@ func (s *QueueStoreContractSuite) TestQueueStore_QueueIsolation() {
 		InFlightCount:   2,
 		Version:         1,
 	}
-	require.NoError(t, s.queueStore.Update(s.ctx, updatedA, 1, 2))
+	require.NoError(t, storeA.Update(s.ctx, updatedA, 1, 2))
 
-	gotB, err := s.queueStore.Get(s.ctx, nameB)
+	gotB, err := storeB.Get(s.ctx, nameB)
 	require.NoError(t, err)
 	assert.Equal(t, baseline, gotB)
+
+	// A store bound to one queue must reject another queue's row outright.
+	require.Error(t, storeA.Create(s.ctx, entity.Queue{Name: nameB, Version: 1}))
+	_, err = storeA.Get(s.ctx, nameB)
+	assert.True(t, storage.IsNotFound(err))
 }