feat(orchestrator): route per-queue storage through host profiles and fix the buildsignal republish dedup

service/submitqueue/orchestrator/server/main.go

@@ -182,7 +182,7 @@ func run() error {
 	// to its own set of extension implementations (conflict analyzer, …),
 	// falling back to a baseline profile for queues without an explicit entry.
 	storageFty := storageFactory{backend: store}
-	profiles, err := newProfiles(logger, scope, changeset.New(storageFty))
+	profiles, err := newProfiles(logger, scope, changeset.New(storageFty), storageFty)
 	if err != nil {
 		return fmt.Errorf("failed to build profiles: %w", err)
 	}
@@ -193,7 +193,7 @@ func run() error {
 	deps := orchestrator.Deps{
 		Logger:         logger.Sugar(),
 		Scope:          scope,
-		Storage:        storageFty,
+		Storage:        profiles.StorageFactory(),
 		Counter:        cnt,
 		BuildRunner:    profiles.BuildRunnerFactory(),
 		ChangeProvider: profiles.ChangeProviderFactory(),

service/submitqueue/orchestrator/server/profiles.go

@@ -27,6 +27,7 @@ import (
 	conflictfake "github.com/uber/submitqueue/submitqueue/extension/conflict/fake"
 	"github.com/uber/submitqueue/submitqueue/extension/conflict/fileoverlap"
 	"github.com/uber/submitqueue/submitqueue/extension/conflict/none"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
 	"go.uber.org/zap"
 )
 
@@ -43,6 +44,11 @@ type Profile struct {
 
 	// Analyzer detects conflicts between concurrent batches in this queue.
 	Analyzer conflict.Analyzer
+
+	// Storage resolves the queue-scoped store aggregate for this queue. Every
+	// profile points at the shared backend by default; a deployment that
+	// splits queues across storage backends overrides this per queue.
+	Storage storage.Factory
 }
 
 // Profiles maps a queue name to its extension Profile, falling back to a
@@ -86,6 +92,14 @@ func (p Profiles) AnalyzerFactory() conflict.Factory {
 	})
 }
 
+// StorageFactory returns a storage.Factory that routes each queue to its
+// profile's storage backend before binding the queue-scoped store aggregate.
+func (p Profiles) StorageFactory() storage.Factory {
+	return storageFunc(func(c storage.Config) (storage.Storage, error) {
+		return p.For(c.QueueName).Storage.For(c)
+	})
+}
+
 // Thin func-type adapters — the http.HandlerFunc trick applied to each
 // extension Factory interface. Each func type satisfies the Factory contract,
 // letting Profiles cross the host/library boundary without dedicated structs.
@@ -104,12 +118,16 @@ type analyzerFunc func(conflict.Config) (conflict.Analyzer, error)
 
 func (f analyzerFunc) For(c conflict.Config) (conflict.Analyzer, error) { return f(c) }
 
+type storageFunc func(storage.Config) (storage.Storage, error)
+
+func (f storageFunc) For(c storage.Config) (storage.Storage, error) { return f(c) }
+
 // newProfiles builds the per-queue extension profiles for the example.
 // Edge integrations (change provider) and the build runner form a shared
 // baseline; each per-queue profile starts from that baseline and overrides
 // only the extensions that differ — here the conflict analyzer.
 // Queues without an explicit profile fall back to the baseline.
-func newProfiles(logger *zap.Logger, scope tally.Scope, resolver changeset.Resolver) (Profiles, error) {
+func newProfiles(logger *zap.Logger, scope tally.Scope, resolver changeset.Resolver, stores storage.Factory) (Profiles, error) {
 	cp, err := newChangeProvider(logger, scope)
 	if err != nil {
 		return Profiles{}, fmt.Errorf("failed to create change provider: %w", err)
@@ -131,6 +149,7 @@ func newProfiles(logger *zap.Logger, scope tally.Scope, resolver changeset.Resol
 		// TODO: replace the delegate with a real analyzer (e.g. Tango target
 		// analysis). "all" serializes the queue conservatively.
 		Analyzer: conflictfake.New(all.New(), nil),
+		Storage:  stores,
 	}
 
 	// e2e-conflict-error-queue: every conflict analysis fails, exercising the

submitqueue/orchestrator/controller/buildsignal/buildsignal.go

@@ -186,7 +186,14 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	// Re-evaluate the batch state machine with the latest build status. The
 	// speculate topic is partitioned by queue like every other speculate
 	// publisher, so a queue's batches keep their serial processing guarantee.
-	if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, updatedBuild.BatchID, batch.Queue); err != nil {
+	// The message ID is scoped to (batch, build): the queue backend dedupes a
+	// publish on (topic, partition, id) against rows not yet garbage-collected,
+	// so reusing the bare batch ID — the ID the batch controller's original
+	// speculate publish used on this same partition — would silently drop this
+	// nudge. A redelivered poll re-mints the same (batch, build) ID, which
+	// dedupes to the first publish, exactly as intended.
+	signalID := fmt.Sprintf("%s/build-signal/%s", updatedBuild.BatchID, updatedBuild.ID)
+	if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, signalID, updatedBuild.BatchID, batch.Queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish to speculate: %w", err)
 	}
@@ -227,15 +234,17 @@ func pollDelay(status entity.BuildStatus) int64 {
 }
 
 // publishBatchID publishes a batch ID to the topic identified by key, stamped
-// with and partitioned by the batch's queue.
-func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, queue string) error {
+// with and partitioned by the batch's queue. msgID is the caller-owned message
+// identity the queue backend dedupes on; it must be distinct from other
+// publishes of the same batch ID on the same partition.
+func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, msgID string, batchID string, queue string) error {
 	bid := entity.BatchID{ID: batchID, Queue: queue}
 	payload, err := bid.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize batch ID: %w", err)
 	}
 
-	msg := entityqueue.NewMessage(batchID, payload, queue, nil)
+	msg := entityqueue.NewMessage(msgID, payload, queue, nil)
 
 	q, ok := c.registry.Queue(key)
 	if !ok {