From 0cc7ace0002d1079bb1e6b49c2c567116efead0c Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Wed, 5 Aug 2026 23:22:50 -0700 Subject: [PATCH] feat(orchestrator): route per-queue storage through host profiles and fix the buildsignal republish dedup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? Per-queue backend routing is host policy: profiles.go already routes every other seam (build runner, change provider, analyzer) per queue, so storage joins the same registry now that its contract is factory-resolved. Separately, end-to-end validation caught a regression from the payload change: repointing buildsignal's speculate publish from the batch-ID partition to the queue partition made its message share the (topic, partition, message-id) triple with the batch controller's original speculate publish, and the MySQL queue backend dedupes publishes on that triple against rows not yet garbage-collected — so the "build finished, re-evaluate" nudge was silently dropped and the batch stalled in Speculating. ### What? `Profile` gains a per-queue storage factory entry; every profile points at the shared MySQL backend by default and `Profiles.StorageFactory()` routes each resolution through the queue's profile, so a deployment that splits queues across backends overrides one field. The orchestrator's `Deps.Storage` now comes from the profile registry. Buildsignal keeps the queue partition (preserving per-queue serial ordering) but mints its speculate message ID from (batch, build) instead of reusing the bare batch ID, so the re-evaluation publish never collides with the batch controller's original message while a redelivered poll still dedupes against itself. ## Test Plan ✅ `make e2e-test` (submitqueue suite green — reproduces the Speculating stall without the message-ID fix) ✅ gateway + orchestrator integration suites via bazel ✅ `go test ./...` ✅ `make fmt` ✅ `make lint` ✅ `make check-tidy` ✅ `make check-gazelle`. Local docker-under-bazel runs need `--sandbox_writable_path=$HOME/.docker` for the image builds. --- .../submitqueue/orchestrator/server/main.go | 4 ++-- .../orchestrator/server/profiles.go | 21 ++++++++++++++++++- .../controller/buildsignal/buildsignal.go | 17 +++++++++++---- 3 files changed, 35 insertions(+), 7 deletions(-) diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index d76ce7a9..2fc2d8e0 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/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(), diff --git a/service/submitqueue/orchestrator/server/profiles.go b/service/submitqueue/orchestrator/server/profiles.go index c1c0f82f..67663d19 100644 --- a/service/submitqueue/orchestrator/server/profiles.go +++ b/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 diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go index 82604c89..5654936e 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go +++ b/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 {