diff --git a/Makefile b/Makefile index d40c5730..335c2883 100644 --- a/Makefile +++ b/Makefile @@ -172,7 +172,7 @@ integration-test-submitqueue-orchestrator: ## Run Orchestrator integration tests license-fix: ## Add missing license headers to source files @$(BAZEL) run //tool/linter/licenseheader -- --fix -lint: lint-fmt lint-license lint-queue-shard ## Run all linters +lint: lint-fmt lint-license lint-message-id lint-queue-shard ## Run all linters @echo "All lint checks passed." lint-fmt: fmt ## Check code formatting (fails if unformatted) @@ -182,6 +182,9 @@ lint-fmt: fmt ## Check code formatting (fails if unformatted) lint-license: ## Check license headers on all source files @$(BAZEL) run //tool/linter/licenseheader -- --check +lint-message-id: ## Check queue messages are only constructed through platform/publish + @$(BAZEL) run //tool/linter/messageid + lint-queue-shard: ## Check every table's primary key leads with the queue column @$(BAZEL) run //tool/linter/queueshard diff --git a/platform/base/messagequeue/message.go b/platform/base/messagequeue/message.go index 7ebd8913..77ef1f00 100644 --- a/platform/base/messagequeue/message.go +++ b/platform/base/messagequeue/message.go @@ -23,6 +23,17 @@ import ( // Immutable - use Copy() for modifications. type Message struct { // ID uniquely identifies the message for deduplication and tracing. + // + // Deduplication is against every message the backend still holds for the + // same topic and partition key — including ones already consumed, which are + // reclaimed lazily and may outlive their delivery by an unbounded interval. + // A publish whose ID collides is reported as a success and stores nothing. + // + // So the ID names the occasion to publish, not the entity published about. + // Reusing an entity's own ID gives that entity one message for as long as + // the backend remembers the first, and silently discards every later one. + // Producers build IDs with platform/publish.IntentID rather than choosing + // them by hand. ID string // Payload is the message body as raw bytes. diff --git a/platform/extension/messagequeue/README.md b/platform/extension/messagequeue/README.md index 128adecf..ab3b3083 100644 --- a/platform/extension/messagequeue/README.md +++ b/platform/extension/messagequeue/README.md @@ -96,10 +96,21 @@ for delivery := range deliveries { } ``` +## Message IDs + +A message ID is the deduplication key, scoped to its topic and partition key. A backend matches a publish against messages it still holds — including ones already consumed, since reclamation is lazy and may lag delivery by an unbounded interval — and a collision is reported to the publisher as a success that stored nothing. There is no error to retry and no row to deliver. + +The ID therefore names the *occasion* to publish, not the entity published about. An entity's own ID buys exactly one message for that entity for as long as the backend remembers the first, so a stage that announces a batch at creation and another that wakes it after a merge would collide, and the wake-up would vanish. + +Producers do not choose IDs by hand. They publish through `platform/publish`, whose `IntentID(entityID, cause...)` composes the entity with the cause of this particular message: a retry of the same cause dedups, which is what makes redelivery safe, while a new cause about the same entity can never be swallowed. `UniqueID` is the fallback for a cause with nothing stable to name it by, and it trades that idempotency for guaranteed delivery. + +Backends must treat the ID as opaque and must not derive routing, ordering, or storage layout from its structure. + ## Implementing a Backend 1. Create `platform/extension/messagequeue/{backend}/` directory 2. Implement `Queue`, `Publisher`, `Subscriber`, `Delivery` interfaces 3. Map `entityqueue.Message` to backend format +4. Deduplicate publishes on (topic, partition key, message ID) See `platform/extension/messagequeue/mysql/` for the reference implementation. diff --git a/submitqueue/core/publish/BUILD.bazel b/platform/publish/BUILD.bazel similarity index 91% rename from submitqueue/core/publish/BUILD.bazel rename to platform/publish/BUILD.bazel index 4e25bee4..1aaf44b6 100644 --- a/submitqueue/core/publish/BUILD.bazel +++ b/platform/publish/BUILD.bazel @@ -3,7 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", srcs = ["publish.go"], - importpath = "github.com/uber/submitqueue/submitqueue/core/publish", + importpath = "github.com/uber/submitqueue/platform/publish", visibility = ["//visibility:public"], deps = [ "//platform/base/messagequeue:go_default_library", diff --git a/platform/publish/publish.go b/platform/publish/publish.go new file mode 100644 index 00000000..5ab6a90e --- /dev/null +++ b/platform/publish/publish.go @@ -0,0 +1,102 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package publish sends a message to the queue behind a topic key. It owns the +// lookup-and-send plumbing every pipeline stage otherwise repeats — resolve the +// key to a queue and a topic name, wrap the payload in a message, publish — and +// the message-ID convention that controls deduplication (see IntentID). +// +// Every producer publishes through this package. Building a message anywhere +// else would put the ID choice back at each call site, which is the mistake the +// convention exists to prevent, so a linter restricts message construction to +// here and to the queue backends. +package publish + +import ( + "context" + "fmt" + "strings" + "sync/atomic" + "time" + + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" +) + +// Message publishes payload to the topic registered for key. +// +// msgID selects the dedup behavior, so the caller must choose it deliberately. +// The queue deduplicates on (topic, partition key, message ID) against every +// row it has not garbage-collected yet, consumed ones included — a window with +// no upper bound on a busy partition. A publish that collides is reported as a +// success and writes nothing, and nothing retries it. +// +// Build msgID with IntentID: name the entity the message is about and the cause +// this particular message exists for. A retry of the same cause then dedups, +// which is what makes redelivery safe, while a new cause about the same entity +// can never be swallowed by an older row. +func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string) error { + q, ok := registry.Queue(key) + if !ok { + return fmt.Errorf("no queue registered for topic key %s", key) + } + topicName, ok := registry.TopicName(key) + if !ok { + return fmt.Errorf("no topic name registered for topic key %s", key) + } + + msg := entityqueue.NewMessage(msgID, payload, partitionKey, nil) + return q.Publisher().Publish(ctx, topicName, msg) +} + +// IntentID names the occasion to publish rather than the entity published +// about: entityID says what the message concerns, and cause says why this +// particular message exists. +// +// Passing no cause asks for at-most-once delivery per entity — every later +// publish about that entity is dropped while an earlier row survives. That is +// right only for a hand-off that happens once in an entity's life, such as +// announcing that it was created. Anything re-sent by design — a wake-up, a +// poll, a re-dispatch, a dead-letter reconciliation — must name its cause, or +// it collides with that one-shot publish and is lost. +// +// Each cause segment must be stable across redeliveries of one occurrence and +// different between occurrences, so derive it from whatever provoked the +// publish: the dependency that reached a terminal state, the build and the +// status observed, the dead letter being reconciled. A wall-clock reading or a +// random value satisfies "different" while destroying "stable", leaving every +// redelivery to publish again. An empty segment carries no information and +// makes two different occasions share an ID, so callers pass none. +func IntentID(entityID string, cause ...string) string { + if len(cause) == 0 { + return entityID + } + return entityID + "/" + strings.Join(cause, "/") +} + +// sequence breaks ties between UniqueID calls that land on the same clock +// tick: some platforms quantize time.Now coarsely enough for consecutive calls +// to read the same nanosecond. +var sequence atomic.Uint64 + +// UniqueID returns a message ID no earlier publish has used, so the publish +// cannot be deduplicated away. +// +// This is the fallback for a cause with nothing stable to name it by, and it +// costs the idempotency IntentID preserves: a redelivery mints a fresh ID and +// publishes a second time, so the consumer has to absorb the duplicate. Prefer +// IntentID wherever the cause can be identified. +func UniqueID(id string) string { + return fmt.Sprintf("%s@%d-%d", id, time.Now().UnixNano(), sequence.Add(1)) +} diff --git a/submitqueue/core/publish/publish_test.go b/platform/publish/publish_test.go similarity index 68% rename from submitqueue/core/publish/publish_test.go rename to platform/publish/publish_test.go index 78bf5cbc..d98f09e9 100644 --- a/submitqueue/core/publish/publish_test.go +++ b/platform/publish/publish_test.go @@ -70,6 +70,47 @@ func TestMessage_UnregisteredKey(t *testing.T) { require.Error(t, err) } +func TestIntentID(t *testing.T) { + tests := []struct { + name string + entityID string + cause []string + want string + }{ + { + name: "no cause is the bare entity ID", + entityID: "batch-1", + want: "batch-1", + }, + { + name: "single cause", + entityID: "batch-1", + cause: []string{"merged"}, + want: "batch-1/merged", + }, + { + name: "multiple causes join in order", + entityID: "batch-1", + cause: []string{"build-signal", "build-9", "running"}, + want: "batch-1/build-signal/build-9/running", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, IntentID(tt.entityID, tt.cause...)) + }) + } +} + +// The convention only works if the same cause is repeatable and a different +// cause is distinguishable — the two properties every call site relies on. +func TestIntentID_StableAcrossCallsAndDistinctPerCause(t *testing.T) { + assert.Equal(t, IntentID("batch-1", "merged"), IntentID("batch-1", "merged")) + assert.NotEqual(t, IntentID("batch-1", "merged"), IntentID("batch-1")) + assert.NotEqual(t, IntentID("batch-1", "merged"), IntentID("batch-1", "cancelling")) +} + func TestUniqueID(t *testing.T) { a := UniqueID("batch-1") b := UniqueID("batch-1") diff --git a/runway/controller/dlq/BUILD.bazel b/runway/controller/dlq/BUILD.bazel index 052b95b9..98668bc3 100644 --- a/runway/controller/dlq/BUILD.bazel +++ b/runway/controller/dlq/BUILD.bazel @@ -8,9 +8,9 @@ go_library( deps = [ "//api/runway/messagequeue:go_default_library", "//api/runway/messagequeue/protopb:go_default_library", - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_zap//:go_default_library", ], diff --git a/runway/controller/dlq/dlq.go b/runway/controller/dlq/dlq.go index c36ed38b..d53f25c7 100644 --- a/runway/controller/dlq/dlq.go +++ b/runway/controller/dlq/dlq.go @@ -43,9 +43,9 @@ import ( "github.com/uber-go/tally" runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" "go.uber.org/zap" ) @@ -153,25 +153,19 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } // publish serializes a MergeResult and publishes it to the signal topic. +// +// Named for the dead letter, because the live handler answers the same request +// on the same topic under the bare correlation ID. Reusing that ID would let +// this terminal failure be deduplicated against an answer that was already +// sent, and the caller would go on waiting for a result nothing will produce. func (c *Controller) publish(ctx context.Context, result *runwaymq.MergeResult, partitionKey string) error { payload, err := runwaymq.Marshal(result) if err != nil { return fmt.Errorf("failed to serialize merge result: %w", err) } - msg := entityqueue.NewMessage(result.GetId(), payload, partitionKey, nil) - - q, ok := c.registry.Queue(c.signalTopicKey) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", c.signalTopicKey) - } - - topicName, ok := c.registry.TopicName(c.signalTopicKey) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", c.signalTopicKey) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + if err := publish.Message(ctx, c.registry, c.signalTopicKey, + publish.IntentID(result.GetId(), "dlq"), payload, partitionKey); err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/runway/controller/merge/BUILD.bazel b/runway/controller/merge/BUILD.bazel index baa9ad86..17434cd1 100644 --- a/runway/controller/merge/BUILD.bazel +++ b/runway/controller/merge/BUILD.bazel @@ -8,9 +8,9 @@ go_library( deps = [ "//api/runway/messagequeue:go_default_library", "//api/runway/messagequeue/protopb:go_default_library", - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//runway/extension/merger:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_zap//:go_default_library", diff --git a/runway/controller/merge/merge.go b/runway/controller/merge/merge.go index 9dc6e074..8a98cb3d 100644 --- a/runway/controller/merge/merge.go +++ b/runway/controller/merge/merge.go @@ -30,9 +30,9 @@ import ( "github.com/uber-go/tally" runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" "github.com/uber/submitqueue/runway/extension/merger" "go.uber.org/zap" ) @@ -141,25 +141,19 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } // publish serializes a MergeResult and publishes it to the given signal topic. +// +// The message ID is the correlation ID with no cause: a request is answered +// once, so a redelivery that re-answers it is meant to dedup rather than tell +// the caller twice. The dead-letter path answers the same request when this +// one never could, and names itself so it cannot be mistaken for a repeat of +// this answer. func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, result *runwaymq.MergeResult, partitionKey string) error { payload, err := runwaymq.Marshal(result) if err != nil { return fmt.Errorf("failed to serialize merge result: %w", err) } - msg := entityqueue.NewMessage(result.GetId(), payload, partitionKey, nil) - - q, ok := c.registry.Queue(key) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", key) - } - - topicName, ok := c.registry.TopicName(key) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", key) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + if err := publish.Message(ctx, c.registry, key, publish.IntentID(result.GetId()), payload, partitionKey); err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/runway/controller/mergeconflictcheck/BUILD.bazel b/runway/controller/mergeconflictcheck/BUILD.bazel index 624e9ff9..dc9ac927 100644 --- a/runway/controller/mergeconflictcheck/BUILD.bazel +++ b/runway/controller/mergeconflictcheck/BUILD.bazel @@ -8,9 +8,9 @@ go_library( deps = [ "//api/runway/messagequeue:go_default_library", "//api/runway/messagequeue/protopb:go_default_library", - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//runway/extension/merger:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_zap//:go_default_library", diff --git a/runway/controller/mergeconflictcheck/mergeconflictcheck.go b/runway/controller/mergeconflictcheck/mergeconflictcheck.go index 87e3d5ba..202f879b 100644 --- a/runway/controller/mergeconflictcheck/mergeconflictcheck.go +++ b/runway/controller/mergeconflictcheck/mergeconflictcheck.go @@ -30,9 +30,9 @@ import ( "github.com/uber-go/tally" runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" "github.com/uber/submitqueue/runway/extension/merger" "go.uber.org/zap" ) @@ -141,25 +141,17 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } // publish serializes a MergeResult and publishes it to the given signal topic. +// +// The message ID is the correlation ID with no cause: a check is answered +// once, so a redelivery that re-answers it is meant to dedup rather than tell +// the caller twice. func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, result *runwaymq.MergeResult, partitionKey string) error { payload, err := runwaymq.Marshal(result) if err != nil { return fmt.Errorf("failed to serialize merge result: %w", err) } - msg := entityqueue.NewMessage(result.GetId(), payload, partitionKey, nil) - - q, ok := c.registry.Queue(key) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", key) - } - - topicName, ok := c.registry.TopicName(key) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", key) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + if err := publish.Message(ctx, c.registry, key, publish.IntentID(result.GetId()), payload, partitionKey); err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/stovepipe/controller/BUILD.bazel b/stovepipe/controller/BUILD.bazel index 5f683870..4bbd1cff 100644 --- a/stovepipe/controller/BUILD.bazel +++ b/stovepipe/controller/BUILD.bazel @@ -10,11 +10,11 @@ go_library( visibility = ["//visibility:public"], deps = [ "//api/stovepipe/protopb:go_default_library", - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/errs:go_default_library", "//platform/extension/counter:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/sourcecontrol:go_default_library", diff --git a/stovepipe/controller/build/BUILD.bazel b/stovepipe/controller/build/BUILD.bazel index 522236a6..ced03370 100644 --- a/stovepipe/controller/build/BUILD.bazel +++ b/stovepipe/controller/build/BUILD.bazel @@ -6,10 +6,10 @@ go_library( importpath = "github.com/uber/submitqueue/stovepipe/controller/build", visibility = ["//visibility:public"], deps = [ - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/errs:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//stovepipe/core/loader:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", diff --git a/stovepipe/controller/build/build.go b/stovepipe/controller/build/build.go index 8da0bc72..8bd08c6c 100644 --- a/stovepipe/controller/build/build.go +++ b/stovepipe/controller/build/build.go @@ -24,10 +24,10 @@ import ( "fmt" "github.com/uber-go/tally" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" "github.com/uber/submitqueue/stovepipe/core/loader" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" @@ -166,23 +166,16 @@ func (c *Controller) loadRequest(ctx context.Context, store storage.Storage, id // publishBuildSignal publishes buildID to the buildsignal stage, partitioned by // build id so each build's poll loop runs in its own partition. +// +// The build ID is the message ID with no cause: a build is handed to the poll +// loop once, so a redelivery that re-hands it off is meant to dedup away. 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) } - msg := entityqueue.NewMessage(buildID, payload, buildID, nil) - - q, ok := c.registry.Queue(stovepipemq.TopicKeyBuildSignal) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", stovepipemq.TopicKeyBuildSignal) - } - topicName, ok := c.registry.TopicName(stovepipemq.TopicKeyBuildSignal) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", stovepipemq.TopicKeyBuildSignal) - } - return q.Publisher().Publish(ctx, topicName, msg) + return publish.Message(ctx, c.registry, stovepipemq.TopicKeyBuildSignal, publish.IntentID(buildID), payload, buildID) } // Name returns the controller name for logging and metrics. diff --git a/stovepipe/controller/buildsignal/BUILD.bazel b/stovepipe/controller/buildsignal/BUILD.bazel index d5b9c17c..bdec7cdf 100644 --- a/stovepipe/controller/buildsignal/BUILD.bazel +++ b/stovepipe/controller/buildsignal/BUILD.bazel @@ -6,9 +6,9 @@ go_library( importpath = "github.com/uber/submitqueue/stovepipe/controller/buildsignal", visibility = ["//visibility:public"], deps = [ - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//stovepipe/core/loader:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", diff --git a/stovepipe/controller/buildsignal/buildsignal.go b/stovepipe/controller/buildsignal/buildsignal.go index e58742c3..c040d239 100644 --- a/stovepipe/controller/buildsignal/buildsignal.go +++ b/stovepipe/controller/buildsignal/buildsignal.go @@ -26,9 +26,9 @@ import ( "fmt" "github.com/uber-go/tally" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" "github.com/uber/submitqueue/stovepipe/core/loader" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" @@ -351,29 +351,15 @@ func pollDelay(status entity.BuildStatus) int64 { } // publishRecord publishes requestID to the record stage, partitioned by request -// 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. +// id. The message id is the request id with no cause, 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, 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) } - msg := entityqueue.NewMessage(requestID, payload, requestID, nil) - return c.publish(ctx, stovepipemq.TopicKeyRecord, msg) -} - -// publish sends msg to the queue registered for key. -func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, msg entityqueue.Message) error { - q, ok := c.registry.Queue(key) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", key) - } - topicName, ok := c.registry.TopicName(key) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", key) - } - return q.Publisher().Publish(ctx, topicName, msg) + return publish.Message(ctx, c.registry, stovepipemq.TopicKeyRecord, publish.IntentID(requestID), payload, requestID) } // Name returns the controller name for logging and metrics. diff --git a/stovepipe/controller/ingest.go b/stovepipe/controller/ingest.go index b8a0b649..68132080 100644 --- a/stovepipe/controller/ingest.go +++ b/stovepipe/controller/ingest.go @@ -20,11 +20,11 @@ import ( "fmt" "github.com/uber-go/tally" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/extension/counter" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" @@ -291,24 +291,16 @@ func (c *IngestController) advanceQueueLatestRequestID(ctx context.Context, stor // publishProcess publishes the request ID to the process stage, partitioned by queue so a // queue's requests stay ordered. +// +// The request ID is the message ID with no cause: a request is handed to +// process once, so a redelivery that re-announces it is meant to dedup away. func (c *IngestController) publishProcess(ctx context.Context, id, queue string) error { payload, err := stovepipemq.Marshal(&stovepipemq.ProcessRequest{Id: id, QueueName: queue}) if err != nil { return fmt.Errorf("failed to serialize process request: %w", err) } - msg := entityqueue.NewMessage(id, payload, queue, nil) - - q, ok := c.registry.Queue(stovepipemq.TopicKeyProcess) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", stovepipemq.TopicKeyProcess) - } - topicName, ok := c.registry.TopicName(stovepipemq.TopicKeyProcess) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", stovepipemq.TopicKeyProcess) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + if err := publish.Message(ctx, c.registry, stovepipemq.TopicKeyProcess, publish.IntentID(id), payload, queue); err != nil { return fmt.Errorf("failed to publish process request: %w", err) } return nil diff --git a/stovepipe/controller/process/BUILD.bazel b/stovepipe/controller/process/BUILD.bazel index ff45ee27..209f87f0 100644 --- a/stovepipe/controller/process/BUILD.bazel +++ b/stovepipe/controller/process/BUILD.bazel @@ -6,10 +6,10 @@ go_library( importpath = "github.com/uber/submitqueue/stovepipe/controller/process", visibility = ["//visibility:public"], deps = [ - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/errs:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//stovepipe/core/loader:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index d4f36bef..0e71fd1c 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -24,10 +24,10 @@ import ( "fmt" "github.com/uber-go/tally" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" "github.com/uber/submitqueue/stovepipe/core/loader" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" @@ -460,21 +460,16 @@ func (c *Controller) loadQueue(ctx context.Context, store storage.Storage, name // 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. +// +// The request ID is the message ID with no cause: a request is admitted to +// build once, so a redelivery that re-admits it is meant to dedup away. 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) } - q, ok := c.registry.Queue(stovepipemq.TopicKeyBuild) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", stovepipemq.TopicKeyBuild) - } - topicName, ok := c.registry.TopicName(stovepipemq.TopicKeyBuild) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", stovepipemq.TopicKeyBuild) - } - if err := q.Publisher().Publish(ctx, topicName, entityqueue.NewMessage(id, payload, id, nil)); err != nil { + if err := publish.Message(ctx, c.registry, stovepipemq.TopicKeyBuild, publish.IntentID(id), payload, id); err != nil { return fmt.Errorf("failed to publish build request: %w", err) } return nil diff --git a/submitqueue/core/publish/publish.go b/submitqueue/core/publish/publish.go deleted file mode 100644 index dd942a2a..00000000 --- a/submitqueue/core/publish/publish.go +++ /dev/null @@ -1,67 +0,0 @@ -// Copyright (c) 2025 Uber Technologies, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -// Package publish sends a message to the queue behind a topic key. It owns the -// lookup-and-send plumbing every orchestrator stage otherwise repeats — resolve -// the key to a queue and a topic name, wrap the payload in a message, publish — -// and the message-ID convention that controls deduplication (see UniqueID). -package publish - -import ( - "context" - "fmt" - "sync/atomic" - "time" - - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" - "github.com/uber/submitqueue/platform/consumer" -) - -// Message publishes payload to the topic registered for key. -// -// msgID selects the dedup behavior, so the caller must choose it deliberately. -// The queue deduplicates on (topic, partition key, message ID) against every -// row it has not garbage-collected yet, consumed ones included: -// -// - A stable msgID (an entity's own ID) makes a repeat publish a silent -// no-op. Right for a hand-off that must happen at most once per entity. -// - UniqueID(id) makes every publish distinct. Right for signals that are -// re-sent by design — wake-ups, polls, re-dispatches — where a swallowed -// repeat would stall the pipeline. -func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string) error { - q, ok := registry.Queue(key) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", key) - } - topicName, ok := registry.TopicName(key) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", key) - } - - msg := entityqueue.NewMessage(msgID, payload, partitionKey, nil) - return q.Publisher().Publish(ctx, topicName, msg) -} - -// sequence breaks ties between UniqueID calls that land on the same clock -// tick: some platforms quantize time.Now coarsely enough for consecutive calls -// to read the same nanosecond. -var sequence atomic.Uint64 - -// UniqueID returns a message ID no earlier publish for the same entity has -// used, so the queue's (topic, partition key, message ID) dedup never swallows -// the repeat. Use it for every publish that is re-sent by design; reusing the -// bare entity ID instead would make the second publish a silent no-op. -func UniqueID(id string) string { - return fmt.Sprintf("%s@%d-%d", id, time.Now().UnixNano(), sequence.Add(1)) -} diff --git a/submitqueue/core/request/BUILD.bazel b/submitqueue/core/request/BUILD.bazel index c47be892..692184c4 100644 --- a/submitqueue/core/request/BUILD.bazel +++ b/submitqueue/core/request/BUILD.bazel @@ -11,8 +11,8 @@ go_library( importpath = "github.com/uber/submitqueue/submitqueue/core/request", visibility = ["//visibility:public"], deps = [ - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/publish:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage:go_default_library", diff --git a/submitqueue/core/request/log.go b/submitqueue/core/request/log.go index b47342b7..598d9a1c 100644 --- a/submitqueue/core/request/log.go +++ b/submitqueue/core/request/log.go @@ -18,8 +18,8 @@ import ( "context" "fmt" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/publish" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" ) @@ -47,23 +47,13 @@ func PublishLog(ctx context.Context, registry consumer.TopicRegistry, logEntry e return fmt.Errorf("failed to serialize request log: %w", err) } - msgID := fmt.Sprintf("%s/%s", logEntry.RequestID, logEntry.Value()) + cause := []string{logEntry.Value()} if occurrence != "" { - msgID = fmt.Sprintf("%s/%s", msgID, occurrence) + cause = append(cause, occurrence) } - msg := entityqueue.NewMessage(msgID, payload, partitionKey, nil) - q, ok := registry.Queue(topickey.TopicKeyLog) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", topickey.TopicKeyLog) - } - - topicName, ok := registry.TopicName(topickey.TopicKeyLog) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", topickey.TopicKeyLog) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + if err := publish.Message(ctx, registry, topickey.TopicKeyLog, + publish.IntentID(logEntry.RequestID, cause...), payload, partitionKey); err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/submitqueue/gateway/controller/BUILD.bazel b/submitqueue/gateway/controller/BUILD.bazel index 5a3f2381..37713674 100644 --- a/submitqueue/gateway/controller/BUILD.bazel +++ b/submitqueue/gateway/controller/BUILD.bazel @@ -15,11 +15,11 @@ go_library( visibility = ["//visibility:public"], deps = [ "//api/submitqueue/gateway/protopb:go_default_library", - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/errs:go_default_library", "//platform/extension/counter:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", diff --git a/submitqueue/gateway/controller/cancel.go b/submitqueue/gateway/controller/cancel.go index d9db08b4..644ae344 100644 --- a/submitqueue/gateway/controller/cancel.go +++ b/submitqueue/gateway/controller/cancel.go @@ -19,10 +19,10 @@ import ( "fmt" "github.com/uber-go/tally" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" requestcore "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -135,20 +135,12 @@ func (c *cancelController) publishToQueue(ctx context.Context, cancelRequest ent } // Partition by the sqid so retries and reorderings on the same request are serialised. - // TODO: figure best way to ID and partition the message according to new guidelines on queue usage - msg := entityqueue.NewMessage(cancelRequest.ID, payload, cancelRequest.ID, nil) - - q, ok := c.registry.Queue(topickey.TopicKeyCancel) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", topickey.TopicKeyCancel) - } - - topicName, ok := c.registry.TopicName(topickey.TopicKeyCancel) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", topickey.TopicKeyCancel) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + // + // The request ID is the message ID with no cause: a request is cancelled at + // most once, so a second Cancel for one already being cancelled is meant to + // dedup rather than enqueue redundant work. + if err := publish.Message(ctx, c.registry, topickey.TopicKeyCancel, + publish.IntentID(cancelRequest.ID), payload, cancelRequest.ID); err != nil { return fmt.Errorf("failed to publish cancel request message: %w", err) } diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go index 0f766b48..0a1a3708 100644 --- a/submitqueue/gateway/controller/land.go +++ b/submitqueue/gateway/controller/land.go @@ -21,11 +21,11 @@ import ( "time" "github.com/uber-go/tally" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/extension/counter" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" requestcore "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -207,23 +207,13 @@ func (c *landController) publishToQueue(ctx context.Context, landRequest entity. return fmt.Errorf("failed to serialize land request: %w", err) } - // Create queue message - // - Message ID: landRequest.ID for idempotency + // Publish the request into the pipeline: + // - Message ID: landRequest.ID with no cause — a request enters once, so a + // retry of this same publish dedups instead of enqueuing it twice // - Payload: serialized LandRequest entity // - Partition key: landRequest.Queue (ensures ordering per queue) - msg := entityqueue.NewMessage(landRequest.ID, payload, landRequest.Queue, nil) - - q, ok := c.registry.Queue(topickey.TopicKeyStart) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", topickey.TopicKeyStart) - } - - topicName, ok := c.registry.TopicName(topickey.TopicKeyStart) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", topickey.TopicKeyStart) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + if err := publish.Message(ctx, c.registry, topickey.TopicKeyStart, + publish.IntentID(landRequest.ID), payload, landRequest.Queue); err != nil { return fmt.Errorf("failed to publish land request message: %w", err) } diff --git a/submitqueue/orchestrator/controller/batch/BUILD.bazel b/submitqueue/orchestrator/controller/batch/BUILD.bazel index 29691b01..cf718405 100644 --- a/submitqueue/orchestrator/controller/batch/BUILD.bazel +++ b/submitqueue/orchestrator/controller/batch/BUILD.bazel @@ -6,10 +6,10 @@ go_library( importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/batch", visibility = ["//visibility:public"], deps = [ - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/extension/counter:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//submitqueue/core/batch:go_default_library", "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", diff --git a/submitqueue/orchestrator/controller/batch/batch.go b/submitqueue/orchestrator/controller/batch/batch.go index f720ec29..0744fa4e 100644 --- a/submitqueue/orchestrator/controller/batch/batch.go +++ b/submitqueue/orchestrator/controller/batch/batch.go @@ -20,10 +20,10 @@ import ( "fmt" "github.com/uber-go/tally" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/extension/counter" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" corebatch "github.com/uber/submitqueue/submitqueue/core/batch" corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" @@ -381,8 +381,14 @@ func (c *Controller) populateBatch(ctx context.Context, store storage.Storage, b return batch, nil } -// publish publishes a batch ID to the specified topic key, stamped with and +// publish announces a batch to the specified topic key, stamped with and // partitioned by the batch's queue. +// +// The message ID is the bare batch ID, with no cause: this is the batch's +// announcement of its own creation, which happens once in its life, so a +// redelivery that re-announces it is meant to be dropped. Every later publish +// about the same batch names its cause and so cannot collide with this row — +// see publish.IntentID. func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID string, queue string) error { bid := entity.BatchID{ID: batchID, Queue: queue} payload, err := bid.ToBytes() @@ -390,19 +396,7 @@ func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID return fmt.Errorf("failed to serialize batch ID: %w", err) } - msg := entityqueue.NewMessage(batchID, payload, queue, nil) - - q, ok := c.registry.Queue(key) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", key) - } - - topicName, ok := c.registry.TopicName(key) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", key) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + if err := publish.Message(ctx, c.registry, key, publish.IntentID(batchID), payload, queue); err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/submitqueue/orchestrator/controller/build/BUILD.bazel b/submitqueue/orchestrator/controller/build/BUILD.bazel index c9d39f3e..ca53a5de 100644 --- a/submitqueue/orchestrator/controller/build/BUILD.bazel +++ b/submitqueue/orchestrator/controller/build/BUILD.bazel @@ -8,7 +8,7 @@ go_library( deps = [ "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", - "//submitqueue/core/publish:go_default_library", + "//platform/publish:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/buildrunner:go_default_library", diff --git a/submitqueue/orchestrator/controller/build/build.go b/submitqueue/orchestrator/controller/build/build.go index bc068673..afa80cc9 100644 --- a/submitqueue/orchestrator/controller/build/build.go +++ b/submitqueue/orchestrator/controller/build/build.go @@ -36,7 +36,7 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" - "github.com/uber/submitqueue/submitqueue/core/publish" + "github.com/uber/submitqueue/platform/publish" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/buildrunner" @@ -371,16 +371,17 @@ func (c *Controller) loadBase(ctx context.Context, store storage.Storage, path e // because they share the key. Partitioning by batch instead would put every // path of a head behind whichever of its builds polls slowest. // -// The build ID is the message ID too — a stable ID on purpose, so a repeat -// hand-off for the same build dedups away while the original signal is still -// in the queue's un-GC'd window (see publish.Message). +// The build ID is the message ID too, with no cause: a build is handed off +// once in its life, so a repeat hand-off for the same build is meant to dedup +// away while the original signal is still in the queue's un-GC'd window (see +// publish.IntentID). func (c *Controller) publishBuildSignal(ctx context.Context, buildID, queue string) error { payload, err := entity.BuildID{ID: buildID, Queue: queue}.ToBytes() if err != nil { return fmt.Errorf("failed to serialize build ID: %w", err) } - if err := publish.Message(ctx, c.registry, topickey.TopicKeyBuildSignal, buildID, payload, buildID); err != nil { + if err := publish.Message(ctx, c.registry, topickey.TopicKeyBuildSignal, publish.IntentID(buildID), payload, buildID); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to publish to buildsignal: %w", err) } diff --git a/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel b/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel index 72e4ee7f..1b5d0e82 100644 --- a/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel +++ b/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel @@ -8,7 +8,7 @@ go_library( deps = [ "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", - "//submitqueue/core/publish:go_default_library", + "//platform/publish:go_default_library", "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go index 2663c508..9c28b869 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go @@ -48,7 +48,7 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" - "github.com/uber/submitqueue/submitqueue/core/publish" + "github.com/uber/submitqueue/platform/publish" corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -244,7 +244,14 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er // Wake the speculate run so it re-plans the queue with this result. It // reads the status from the record above rather than being told it, so a // duplicated or reordered signal costs nothing. - if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, batch.ID, batch.Queue); err != nil { + // + // The cause is this build at this status, so every transition wakes + // speculate exactly once and the polls in between — which observed nothing + // new — collapse into the wake-up already sent. Naming only the build would + // dedup the terminal wake-up against the very first poll's. + if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, + publish.IntentID(batch.ID, "build-signal", build.ID, string(status)), + batch.ID, batch.Queue); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to publish to speculate: %w", err) } @@ -426,16 +433,14 @@ func findEntry(set entity.SpeculationPathSet, pathID string) (entity.Speculation return entity.SpeculationPathEntry{}, false } -// publishBatchID publishes a batch ID to the topic identified by key, stamped -// with and partitioned by the batch's queue, with a distinct message ID per -// publish (publish.UniqueID) so a later wake-up for the same batch is never -// deduplicated away. -func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, queue string) error { +// publishBatchID publishes a batch ID to the topic identified by key under +// msgID, stamped with and partitioned by the batch's queue. +func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, msgID, batchID, queue string) error { payload, err := entity.BatchID{ID: batchID, Queue: queue}.ToBytes() if err != nil { return fmt.Errorf("failed to serialize batch ID: %w", err) } - return publish.Message(ctx, c.registry, key, publish.UniqueID(batchID), payload, queue) + return publish.Message(ctx, c.registry, key, msgID, payload, queue) } // Name returns the controller name for logging and metrics. diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go index 6965d382..9e02cda1 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go @@ -229,6 +229,48 @@ func TestProcess_RecordsStatusAndNeverWritesThePathSet(t *testing.T) { } } +// The wake-up this poll sends speculate is named for the status it observed. +// +// Every poll publishes, but only a status change carries news, and the queue +// deduplicates on (topic, partition key, message ID) against rows it has not +// collected yet — so polls that saw nothing new collapse into the wake-up +// already sent, while each transition gets one of its own. Naming only the +// build would collapse the terminal wake-up into the first poll's and strand +// the batch; naming nothing stable would wake speculate on every tick of a +// build that has not moved. +func TestProcess_SpeculateWakeUpIsNamedForTheObservedStatus(t *testing.T) { + poll := func(t *testing.T, from, to entity.BuildStatus) string { + t.Helper() + ctrl := gomock.NewController(t) + h := newTestHarness(t, ctrl, entity.BatchStateSpeculating) + h.wanted() + + h.builds.EXPECT().Get(gomock.Any(), testBuildID).Return(testBuild(from), nil) + h.br.EXPECT().Status(gomock.Any(), gomock.Any()).Return(to, nil, nil) + h.builds.EXPECT().Update(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + + var id string + h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + id = msg.ID + return nil + }, + ) + d := delivery(t, ctrl) + d.EXPECT().Hold(gomock.Any()).AnyTimes() + + require.NoError(t, h.controller.Process(context.Background(), d)) + return id + } + + running := poll(t, entity.BuildStatusRunning, entity.BuildStatusRunning) + assert.Equal(t, running, poll(t, entity.BuildStatusRunning, entity.BuildStatusRunning), + "two polls that observed the same status carry the same news") + assert.NotEqual(t, running, poll(t, entity.BuildStatusRunning, entity.BuildStatusSucceeded), + "the terminal wake-up must not be swallowed by an earlier poll's") + assert.NotEqual(t, testBatchID, running, "the batch announcement already holds the bare ID") +} + // While a build is in flight the loop holds the delivery for the poll delay, // so the same message — and the same build partition — carries every poll. func TestProcess_NonTerminalHoldsForNextPoll(t *testing.T) { diff --git a/submitqueue/orchestrator/controller/cancel/BUILD.bazel b/submitqueue/orchestrator/controller/cancel/BUILD.bazel index 03a3b557..df905bd8 100644 --- a/submitqueue/orchestrator/controller/cancel/BUILD.bazel +++ b/submitqueue/orchestrator/controller/cancel/BUILD.bazel @@ -8,8 +8,8 @@ go_library( deps = [ "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//submitqueue/core/batch:go_default_library", - "//submitqueue/core/publish:go_default_library", "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", diff --git a/submitqueue/orchestrator/controller/cancel/cancel.go b/submitqueue/orchestrator/controller/cancel/cancel.go index b78b04c3..aad8e0c5 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel.go +++ b/submitqueue/orchestrator/controller/cancel/cancel.go @@ -59,8 +59,8 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" corebatch "github.com/uber/submitqueue/submitqueue/core/batch" - "github.com/uber/submitqueue/submitqueue/core/publish" corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" diff --git a/submitqueue/orchestrator/controller/dlq/BUILD.bazel b/submitqueue/orchestrator/controller/dlq/BUILD.bazel index c8840e5f..ffcbd2eb 100644 --- a/submitqueue/orchestrator/controller/dlq/BUILD.bazel +++ b/submitqueue/orchestrator/controller/dlq/BUILD.bazel @@ -18,8 +18,8 @@ go_library( "//api/runway/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//submitqueue/core/batch:go_default_library", - "//submitqueue/core/publish:go_default_library", "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", diff --git a/submitqueue/orchestrator/controller/dlq/speculate.go b/submitqueue/orchestrator/controller/dlq/speculate.go index 3c41c6b5..7e6fe587 100644 --- a/submitqueue/orchestrator/controller/dlq/speculate.go +++ b/submitqueue/orchestrator/controller/dlq/speculate.go @@ -21,8 +21,8 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" corebatch "github.com/uber/submitqueue/submitqueue/core/batch" - "github.com/uber/submitqueue/submitqueue/core/publish" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" diff --git a/submitqueue/orchestrator/controller/merge/BUILD.bazel b/submitqueue/orchestrator/controller/merge/BUILD.bazel index 6370a4fc..a0eb8359 100644 --- a/submitqueue/orchestrator/controller/merge/BUILD.bazel +++ b/submitqueue/orchestrator/controller/merge/BUILD.bazel @@ -10,9 +10,9 @@ go_library( "//api/base/mergestrategy/protopb:go_default_library", "//api/runway/messagequeue:go_default_library", "//platform/base/mergestrategy:go_default_library", - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//submitqueue/core/request:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage:go_default_library", diff --git a/submitqueue/orchestrator/controller/merge/merge.go b/submitqueue/orchestrator/controller/merge/merge.go index 0fa9c042..a9657457 100644 --- a/submitqueue/orchestrator/controller/merge/merge.go +++ b/submitqueue/orchestrator/controller/merge/merge.go @@ -32,9 +32,9 @@ import ( strategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" "github.com/uber/submitqueue/platform/base/mergestrategy" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" @@ -219,25 +219,17 @@ func toProtoStrategy(s mergestrategy.MergeStrategy) strategypb.Strategy { // publish serializes the runway merge request and publishes it to the given // topic key, partitioned by queue. +// +// The correlation ID is the message ID with no cause: a batch is asked to merge +// once, so a redelivery that re-asks is meant to dedup rather than have Runway +// merge the same batch twice. func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, req *runwaymq.MergeRequest, partitionKey string) error { payload, err := runwaymq.Marshal(req) if err != nil { return fmt.Errorf("failed to serialize merge request: %w", err) } - msg := entityqueue.NewMessage(req.Id, payload, partitionKey, nil) - - q, ok := c.registry.Queue(key) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", key) - } - - topicName, ok := c.registry.TopicName(key) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", key) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + if err := publish.Message(ctx, c.registry, key, publish.IntentID(req.GetId()), payload, partitionKey); err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/submitqueue/orchestrator/controller/mergeconflictsignal/BUILD.bazel b/submitqueue/orchestrator/controller/mergeconflictsignal/BUILD.bazel index 07ac5501..94f0c282 100644 --- a/submitqueue/orchestrator/controller/mergeconflictsignal/BUILD.bazel +++ b/submitqueue/orchestrator/controller/mergeconflictsignal/BUILD.bazel @@ -8,9 +8,9 @@ go_library( deps = [ "//api/runway/messagequeue:go_default_library", "//api/runway/messagequeue/protopb:go_default_library", - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", diff --git a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go index 2402f601..dbfb69b5 100644 --- a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go +++ b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go @@ -26,9 +26,9 @@ import ( "github.com/uber-go/tally" runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -197,25 +197,17 @@ func (c *Controller) failRequest(ctx context.Context, store storage.Storage, req // publishRequestID publishes a request ID to the given topic key, stamped with // and partitioned by the request's queue. +// +// The request ID is the message ID with no cause: a request is handed on once +// per conflict-check result, so a redelivery that re-hands it is meant to dedup +// away. func (c *Controller) publishRequestID(ctx context.Context, key consumer.TopicKey, requestID string, queue string) error { payload, err := entity.RequestID{ID: requestID, Queue: queue}.ToBytes() if err != nil { return fmt.Errorf("failed to serialize request ID: %w", err) } - msg := entityqueue.NewMessage(requestID, payload, queue, nil) - - q, ok := c.registry.Queue(key) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", key) - } - - topicName, ok := c.registry.TopicName(key) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", key) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + if err := publish.Message(ctx, c.registry, key, publish.IntentID(requestID), payload, queue); err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/submitqueue/orchestrator/controller/mergesignal/BUILD.bazel b/submitqueue/orchestrator/controller/mergesignal/BUILD.bazel index aa85d1aa..9163af3d 100644 --- a/submitqueue/orchestrator/controller/mergesignal/BUILD.bazel +++ b/submitqueue/orchestrator/controller/mergesignal/BUILD.bazel @@ -8,9 +8,9 @@ go_library( deps = [ "//api/runway/messagequeue:go_default_library", "//api/runway/messagequeue/protopb:go_default_library", - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//submitqueue/core/batch:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", diff --git a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go index 6d94b7f7..795c1d99 100644 --- a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go +++ b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go @@ -28,9 +28,9 @@ import ( "github.com/uber-go/tally" runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" corebatch "github.com/uber/submitqueue/submitqueue/core/batch" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -162,39 +162,35 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er // fanout publishes the batch ID to conclude (so requests are updated) and to // speculate (so dependents can re-evaluate now that this batch is done). +// +// Both messages name the merge as their cause. Without it the speculate +// publish would reuse the bare batch ID, which the batch controller already +// published at creation, and the queue would drop this one as a duplicate for +// as long as that row survives — leaving dependents unwoken. Conclude is +// scoped the same way because speculate publishes there too when a batch goes +// terminal on its own; the two mean the same thing but are decided at +// different moments, so neither may swallow the other. func (c *Controller) fanout(ctx context.Context, batchID, queue string) error { - if err := c.publish(ctx, topickey.TopicKeyConclude, batchID, queue); err != nil { + if err := c.publish(ctx, topickey.TopicKeyConclude, publish.IntentID(batchID, "conclude", "merged"), batchID, queue); err != nil { metrics.NamedCounter(c.metricsScope, "process", "publish_conclude_errors", 1) return fmt.Errorf("failed to publish to conclude: %w", err) } - if err := c.publish(ctx, topickey.TopicKeySpeculate, batchID, queue); err != nil { + if err := c.publish(ctx, topickey.TopicKeySpeculate, publish.IntentID(batchID, "merged"), batchID, queue); err != nil { metrics.NamedCounter(c.metricsScope, "process", "publish_speculate_errors", 1) return fmt.Errorf("failed to publish to speculate: %w", err) } return nil } -// publish publishes a batch ID to the given topic key, stamped with and -// partitioned by the batch's queue. -func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID string, queue string) error { +// publish publishes a batch ID to the given topic key under msgID, stamped +// with and partitioned by the batch's queue. +func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, msgID, batchID, queue string) error { payload, err := entity.BatchID{ID: batchID, Queue: queue}.ToBytes() if err != nil { return fmt.Errorf("failed to serialize batch ID: %w", err) } - msg := entityqueue.NewMessage(batchID, payload, queue, nil) - - q, ok := c.registry.Queue(key) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", key) - } - - topicName, ok := c.registry.TopicName(key) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", key) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + if err := publish.Message(ctx, c.registry, key, msgID, payload, queue); err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go b/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go index 8655ba31..2519c5b8 100644 --- a/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go +++ b/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go @@ -152,6 +152,59 @@ func TestProcess_MergedAdvancesBatch(t *testing.T) { assert.ElementsMatch(t, []string{"conclude", "speculate"}, got) } +// The fan-out after a merge must not reuse the bare batch ID. +// +// The batch controller announces a new batch to speculate under exactly that +// ID, and the queue deduplicates on (topic, partition key, message ID) against +// every row it has not collected yet, consumed ones included — a window with no +// upper bound on a busy partition. Reusing the ID here made the wake-up that +// lets dependents re-plan a silent no-op, acked as a success, with nothing to +// retry it. +func TestProcess_FanoutDoesNotCollideWithTheBatchAnnouncement(t *testing.T) { + ctrl := gomock.NewController(t) + + batchStore := storagemock.NewMockBatchStore(ctrl) + batch := entity.Batch{ + ID: testBatchID, + Queue: testQueue, + Contains: []string{"test-queue/1"}, + Dependencies: []string{"test-queue/batch/0"}, + State: entity.BatchStateMerging, + Version: 1, + } + batchStore.EXPECT().Get(gomock.Any(), testBatchID).Return(batch, nil) + batchStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil) + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + + byTopic := map[string]string{} + pub := queuemock.NewMockPublisher(ctrl) + pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, topic string, msg entityqueue.Message) error { + byTopic[topic] = msg.ID + return nil + }, + ).AnyTimes() + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(pub).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: q}, + {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: q}, + }) + require.NoError(t, err) + + res := runwaymq.MergeResult{Id: testBatchID, Outcome: runwaypb.Outcome_SUCCEEDED} + msg := entityqueue.NewMessage(testBatchID, resultPayload(t, res), testQueue, nil) + require.NoError(t, newController(t, store, registry).Process(context.Background(), newDelivery(ctrl, msg))) + + assert.NotEqual(t, testBatchID, byTopic["speculate"], + "the announcement the batch controller published already holds this ID") + assert.NotEqual(t, testBatchID, byTopic["conclude"]) + assert.NotEqual(t, byTopic["speculate"], byTopic["conclude"]) +} + func TestProcess_NotMergedMarksBatchFailed(t *testing.T) { ctrl := gomock.NewController(t) diff --git a/submitqueue/orchestrator/controller/speculate/BUILD.bazel b/submitqueue/orchestrator/controller/speculate/BUILD.bazel index fe47b2c8..01b4113a 100644 --- a/submitqueue/orchestrator/controller/speculate/BUILD.bazel +++ b/submitqueue/orchestrator/controller/speculate/BUILD.bazel @@ -19,8 +19,8 @@ go_library( "//platform/consumer:go_default_library", "//platform/errs:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//submitqueue/core/batch:go_default_library", - "//submitqueue/core/publish:go_default_library", "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", diff --git a/submitqueue/orchestrator/controller/speculate/dispatch.go b/submitqueue/orchestrator/controller/speculate/dispatch.go index 1732eca5..c489ace9 100644 --- a/submitqueue/orchestrator/controller/speculate/dispatch.go +++ b/submitqueue/orchestrator/controller/speculate/dispatch.go @@ -21,6 +21,7 @@ import ( "time" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" @@ -99,8 +100,13 @@ func (c *Controller) dispatch(ctx context.Context, queue string, snap snapshot, // Cancelling paths need no dispatch at all — the poll loop reads the // stop off the set and enacts it. Partitioned by batch, so heads // dispatch in parallel while one head's dispatches stay ordered. + // + // Distinct per publish: the condition that provokes a re-send is that + // nothing recorded the last one, which leaves nothing new to name it + // by, so any stable ID would dedup the re-send against the dispatch + // that went missing. if hasActionablePaths(set) { - if err := c.publishBatchID(ctx, topickey.TopicKeyBuild, batch.ID, queue, batch.ID); err != nil { + if err := c.publishBatchID(ctx, topickey.TopicKeyBuild, publish.UniqueID(batch.ID), batch.ID, queue, batch.ID); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return c.attributed(fmt.Errorf("failed to publish batch %s to build: %w", batch.ID, err), entity.BatchSubject(batch.ID)) diff --git a/submitqueue/orchestrator/controller/speculate/finalize.go b/submitqueue/orchestrator/controller/speculate/finalize.go index 04cd5256..98d210cf 100644 --- a/submitqueue/orchestrator/controller/speculate/finalize.go +++ b/submitqueue/orchestrator/controller/speculate/finalize.go @@ -21,6 +21,7 @@ import ( "time" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" corebatch "github.com/uber/submitqueue/submitqueue/core/batch" corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" @@ -321,7 +322,11 @@ func (c *Controller) applyOutcome(ctx context.Context, store storage.Storage, ba switch decision { case outcomeMerge: state = entity.BatchStateMerging - if err := c.publishBatchID(ctx, topickey.TopicKeyMerge, batch.ID, batch.Queue, batch.Queue); err != nil { + // A batch merges once, so the dispatch names only that as its cause: a + // redelivery that re-derives outcomeMerge because the state write was + // lost dedups against the request already sent, instead of asking + // Runway to merge the same batch twice. + if err := c.publishBatchID(ctx, topickey.TopicKeyMerge, publish.IntentID(batch.ID, "merge-dispatch"), batch.ID, batch.Queue, batch.Queue); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return false, fmt.Errorf("failed to publish batch %s to merge: %w", batch.ID, err) } @@ -365,7 +370,12 @@ func (c *Controller) applyOutcome(ctx context.Context, store storage.Storage, ba ) if terminal { - if err := c.publishBatchID(ctx, topickey.TopicKeyConclude, batch.ID, batch.Queue, batch.Queue); err != nil { + // Named for the run that decided it, so a redelivery re-deriving the + // same outcome does not conclude the batch twice, and so it stays + // distinct from the conclude mergesignal sends for a merged batch. + // A conclude that goes missing is recovered by fanout, which is + // deliberately un-deduplicated. + if err := c.publishBatchID(ctx, topickey.TopicKeyConclude, publish.IntentID(batch.ID, "conclude", "speculate"), batch.ID, batch.Queue, batch.Queue); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return true, fmt.Errorf("failed to publish batch %s to conclude: %w", batch.ID, err) } @@ -383,6 +393,10 @@ func (c *Controller) applyOutcome(ctx context.Context, store storage.Storage, ba // message, and once terminal it is gone from the queue listing — so without // this its requests would simply stay unreconciled. // +// Distinct per publish: the guarantee being bought is that a message exists at +// all, and a stable ID would let the queue answer "one already did" with a +// success that writes nothing. +// // Published before the state write, not after: sent afterwards it is one more // thing that can fail exactly when everything else is failing, leaving the // obligation created and the means to discharge it gone. Sent first, a @@ -391,7 +405,7 @@ func (c *Controller) applyOutcome(ctx context.Context, store storage.Storage, ba // tolerates any batch state, and if the write never lands the message is just // a nudge that re-plans a queue nothing has changed. func (c *Controller) recoverable(ctx context.Context, store storage.Storage, batch entity.Batch) error { - if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, batch.ID, batch.Queue, batch.Queue); err != nil { + if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, publish.UniqueID(batch.ID), batch.ID, batch.Queue, batch.Queue); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to publish recovery signal for batch %s: %w", batch.ID, err) } diff --git a/submitqueue/orchestrator/controller/speculate/run_test.go b/submitqueue/orchestrator/controller/speculate/run_test.go index 596a51b5..aa1fe896 100644 --- a/submitqueue/orchestrator/controller/speculate/run_test.go +++ b/submitqueue/orchestrator/controller/speculate/run_test.go @@ -1013,11 +1013,12 @@ func TestRun_PersistsObservationsWithNoOpenHead(t *testing.T) { h.noBuildsDispatched() } -// Repeat publishes for one batch must reach the queue. It deduplicates on -// (topic, partition key, message ID) against rows it has not collected yet, -// consumed ones included, so a bare batch ID would silently drop the re-sends -// this controller relies on. -func TestPublish_MintsADistinctMessageIDPerPublish(t *testing.T) { +// The self-heal fan-out exists because an earlier conclude may have gone +// missing, so each of its publishes must reach the queue. The queue +// deduplicates on (topic, partition key, message ID) against rows it has not +// collected yet, consumed ones included, so a stable ID would drop the repeat +// against the very conclude that went missing. +func TestFanout_MintsADistinctMessageIDPerPublish(t *testing.T) { ctrl := gomock.NewController(t) var ids []string @@ -1041,8 +1042,8 @@ func TestPublish_MintsADistinctMessageIDPerPublish(t *testing.T) { staticSpeculatorFactory{}, registry, topickey.TopicKeySpeculate, "orchestrator-speculate", ) - require.NoError(t, c.publishBatchID(context.Background(), topickey.TopicKeyConclude, head, "q", "q")) - require.NoError(t, c.publishBatchID(context.Background(), topickey.TopicKeyConclude, head, "q", "q")) + require.NoError(t, c.fanout(context.Background(), head, "q")) + require.NoError(t, c.fanout(context.Background(), head, "q")) require.Len(t, ids, 2) assert.NotEqual(t, ids[0], ids[1]) diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go index ee819a4d..9ce4782a 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate.go +++ b/submitqueue/orchestrator/controller/speculate/speculate.go @@ -23,8 +23,8 @@ import ( "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" corebatch "github.com/uber/submitqueue/submitqueue/core/batch" - "github.com/uber/submitqueue/submitqueue/core/publish" corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -182,8 +182,12 @@ func (c *Controller) admit(ctx context.Context, store storage.Storage, batch ent // fanout re-publishes downstream events for a batch that has already reached // a terminal state. Used for self-healing when a previous publish was lost: // re-sending to conclude guarantees request-state reconciliation. +// +// Distinct per publish: this exists precisely because an earlier conclude may +// have gone missing, and the conclude the batch sent when it turned terminal +// is exactly what a stable ID would dedup against. func (c *Controller) fanout(ctx context.Context, batchID, queue string) error { - if err := c.publishBatchID(ctx, topickey.TopicKeyConclude, batchID, queue, queue); err != nil { + if err := c.publishBatchID(ctx, topickey.TopicKeyConclude, publish.UniqueID(batchID), batchID, queue, queue); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return c.attributed(fmt.Errorf("failed to publish to conclude: %w", err), entity.BatchSubject(batchID)) @@ -191,13 +195,18 @@ func (c *Controller) fanout(ctx context.Context, batchID, queue string) error { return nil } -// publishBatchID publishes a batch ID to the topic behind key, stamped with -// the batch's queue and partitioned by partitionKey. +// publishBatchID publishes a batch ID to the topic behind key under msgID, +// stamped with the batch's queue and partitioned by partitionKey. // -// Every publish gets a distinct message ID (publish.UniqueID): this controller -// re-publishes by design — a dispatch is re-sent until the build stage records -// it, and a terminal batch repeats its fan-out in case an earlier one was lost -// — and a stable message ID would make the queue swallow those repeats. +// Callers choose msgID, because this controller publishes for two different +// kinds of reason. A hand-off that happens once in a batch's life — dispatching +// it to merge, concluding it — names its cause with publish.IntentID, so a +// redelivery that re-derives the same decision is deduplicated instead of +// enacting it twice. A repeat-until-effective nudge — a dispatch re-sent until +// the build stage records it, a fan-out repeated in case an earlier one was +// lost — has no stable cause to name, since the condition that provokes it is +// unchanged across runs, and takes publish.UniqueID so the queue can never +// swallow the repeat that finally lands. // // queue and partitionKey are separate because they answer different questions. // queue is what the payload asserts the batch belongs to, and the consumer @@ -205,12 +214,12 @@ func (c *Controller) fanout(ctx context.Context, batchID, queue string) error { // partition key only decides what stays serialized behind what: most publishes // want the queue, so a queue's batches are processed in order, but the build // dispatch partitions by batch so heads dispatch in parallel. -func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID, queue, partitionKey string) error { +func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, msgID, batchID, queue, partitionKey string) error { payload, err := entity.BatchID{ID: batchID, Queue: queue}.ToBytes() if err != nil { return fmt.Errorf("failed to serialize batch ID: %w", err) } - return publish.Message(ctx, c.registry, key, publish.UniqueID(batchID), payload, partitionKey) + return publish.Message(ctx, c.registry, key, msgID, payload, partitionKey) } // attributed records what a failure was about and counts it by subject type. diff --git a/submitqueue/orchestrator/controller/start/BUILD.bazel b/submitqueue/orchestrator/controller/start/BUILD.bazel index 960d163d..b86bcd76 100644 --- a/submitqueue/orchestrator/controller/start/BUILD.bazel +++ b/submitqueue/orchestrator/controller/start/BUILD.bazel @@ -6,9 +6,9 @@ go_library( importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/start", visibility = ["//visibility:public"], deps = [ - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", diff --git a/submitqueue/orchestrator/controller/start/start.go b/submitqueue/orchestrator/controller/start/start.go index 14555db6..5dbfbd04 100644 --- a/submitqueue/orchestrator/controller/start/start.go +++ b/submitqueue/orchestrator/controller/start/start.go @@ -20,9 +20,9 @@ import ( "fmt" "github.com/uber-go/tally" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -139,6 +139,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er // publish publishes a request ID to the specified topic key, stamped with and // partitioned by the request's queue. +// +// The request ID is the message ID with no cause: a request is handed to the +// next stage once, so a redelivery that re-hands it is meant to dedup away. func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, requestID string, queue string) error { rid := entity.RequestID{ID: requestID, Queue: queue} payload, err := rid.ToBytes() @@ -146,19 +149,7 @@ func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, request return fmt.Errorf("failed to serialize request ID: %w", err) } - msg := entityqueue.NewMessage(requestID, payload, queue, nil) - - q, ok := c.registry.Queue(key) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", key) - } - - topicName, ok := c.registry.TopicName(key) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", key) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + if err := publish.Message(ctx, c.registry, key, publish.IntentID(requestID), payload, queue); err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/submitqueue/orchestrator/controller/validate/BUILD.bazel b/submitqueue/orchestrator/controller/validate/BUILD.bazel index 8f2a923f..af59d425 100644 --- a/submitqueue/orchestrator/controller/validate/BUILD.bazel +++ b/submitqueue/orchestrator/controller/validate/BUILD.bazel @@ -10,9 +10,9 @@ go_library( "//api/base/mergestrategy/protopb:go_default_library", "//api/runway/messagequeue:go_default_library", "//platform/base/mergestrategy:go_default_library", - "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//platform/publish:go_default_library", "//submitqueue/core/request:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/changeprovider:go_default_library", diff --git a/submitqueue/orchestrator/controller/validate/validate.go b/submitqueue/orchestrator/controller/validate/validate.go index 3e3d7ce2..9eacab3c 100644 --- a/submitqueue/orchestrator/controller/validate/validate.go +++ b/submitqueue/orchestrator/controller/validate/validate.go @@ -25,9 +25,9 @@ import ( strategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" "github.com/uber/submitqueue/platform/base/mergestrategy" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" coremetrics "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/platform/publish" corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/changeprovider" @@ -311,25 +311,17 @@ func (c *Controller) checkDuplicate(ctx context.Context, store storage.Storage, // publishMergeCheck serializes the runway check request and publishes it to the // runway merge-conflict-check topic, partitioned by queue. +// +// The correlation ID is the message ID with no cause: a request is checked once, +// so a redelivery that re-asks is meant to dedup rather than have Runway run the +// same check twice. func (c *Controller) publishMergeCheck(ctx context.Context, req *runwaymq.MergeRequest) error { payload, err := runwaymq.Marshal(req) if err != nil { return fmt.Errorf("failed to serialize merge conflict check request: %w", err) } - msg := entityqueue.NewMessage(req.Id, payload, req.QueueName, nil) - - q, ok := c.registry.Queue(c.runwayTopicKey) - if !ok { - return fmt.Errorf("no queue registered for topic key %s", c.runwayTopicKey) - } - - topicName, ok := c.registry.TopicName(c.runwayTopicKey) - if !ok { - return fmt.Errorf("no topic name registered for topic key %s", c.runwayTopicKey) - } - - if err := q.Publisher().Publish(ctx, topicName, msg); err != nil { + if err := publish.Message(ctx, c.registry, c.runwayTopicKey, publish.IntentID(req.GetId()), payload, req.GetQueueName()); err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/test/integration/extension/messagequeue/mysql/queue_test.go b/test/integration/extension/messagequeue/mysql/queue_test.go index 65de942c..7faf659f 100644 --- a/test/integration/extension/messagequeue/mysql/queue_test.go +++ b/test/integration/extension/messagequeue/mysql/queue_test.go @@ -699,6 +699,58 @@ func (s *SQLQueueIntegrationSuite) TestIdempotentPublish() { t.Logf("Confirmed: only received message once (idempotency works)") } +// A consumed and acked message goes on deduplicating later publishes. +// +// This is the property the message-ID convention exists for, and the one that +// surprises: acked rows are reclaimed lazily, so a publish can be swallowed by +// a message that was already delivered — reported to the publisher as a success +// that writes nothing, with no error to retry and no row to deliver. A producer +// reusing an entity's ID therefore gets one message per entity for as long as +// the backend remembers the first. See platform/publish.IntentID. +func (s *SQLQueueIntegrationSuite) TestDedupOutlivesConsumption() { + t := s.T() + + signalCh := make(chan queueMySQL.HookSignal, 100) + q, err := queueMySQL.NewQueue(queueMySQL.Params{ + DB: s.db, + Logger: zaptest.NewLogger(t), + MetricsScope: tally.NoopScope, + OnSignal: signalCh, + }) + require.NoError(t, err) + defer q.Close() + + publisher, subscriber := q.Publisher(), q.Subscriber() + topic := "dedup_after_ack_topic" + + subConfig := extqueue.DefaultSubscriptionConfig("worker-1", "dedup-after-ack-consumer") + subConfig.PartitionDiscoveryIntervalMs = 100 + deliveryChan, err := subscriber.Subscribe(s.ctx, topic, subConfig) + require.NoError(t, err) + + // The first publish for an entity: delivered, acked, and now awaiting + // collection rather than gone. + require.NoError(t, publisher.Publish(s.ctx, topic, + entityqueue.NewMessage("batch-1", []byte("announced"), "queue-1", nil))) + first := receive(t, deliveryChan) + require.Equal(t, "batch-1", first.Message().ID) + require.NoError(t, first.Ack(s.ctx)) + + // A later, unrelated event about the same entity, published under the same + // ID. It reports success and is never delivered. + require.NoError(t, publisher.Publish(s.ctx, topic, + entityqueue.NewMessage("batch-1", []byte("woken"), "queue-1", nil))) + assertNoDelivery(t, deliveryChan, signalCh, queueMySQL.SignalDeliveryCheck, 3) + + // Naming the cause is what gets it through. + require.NoError(t, publisher.Publish(s.ctx, topic, + entityqueue.NewMessage("batch-1/merged", []byte("woken"), "queue-1", nil))) + second := receive(t, deliveryChan) + assert.Equal(t, "batch-1/merged", second.Message().ID) + assert.Equal(t, []byte("woken"), second.Message().Payload) + require.NoError(t, second.Ack(s.ctx)) +} + func (s *SQLQueueIntegrationSuite) TestConcurrentPublishers() { t := s.T() diff --git a/tool/linter/messageid/BUILD.bazel b/tool/linter/messageid/BUILD.bazel new file mode 100644 index 00000000..204bb100 --- /dev/null +++ b/tool/linter/messageid/BUILD.bazel @@ -0,0 +1,24 @@ +load("@rules_go//go:def.bzl", "go_binary", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["main.go"], + importpath = "github.com/uber/submitqueue/tool/linter/messageid", + visibility = ["//visibility:private"], +) + +go_binary( + name = "messageid", + embed = [":go_default_library"], + visibility = ["//visibility:public"], +) + +go_test( + name = "go_default_test", + srcs = ["main_test.go"], + embed = [":go_default_library"], + deps = [ + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/tool/linter/messageid/main.go b/tool/linter/messageid/main.go new file mode 100644 index 00000000..c633c9bb --- /dev/null +++ b/tool/linter/messageid/main.go @@ -0,0 +1,228 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Command messageid checks that queue messages are only constructed where the +// message-ID convention is enforced. +// +// A message ID is a deduplication key, matched against every message the +// backend still holds for the same topic and partition key — consumed ones +// included. A publish whose ID collides is reported as a success and stores +// nothing, so an ID chosen carelessly does not fail loudly; it drops an event. +// The safe choice is not something a reviewer can see locally either, because +// it depends on what every other producer on that topic uses. +// +// Whether a given ID expression is well chosen cannot be decided by reading it, +// so this does not try. It enforces the structural rule that makes the question +// answerable in one place: platform/base/messagequeue.NewMessage may only be +// called from platform/publish, which composes IDs from an entity and a cause, +// and from the queue backends, which rebuild messages they are handed back. +// Every producer then goes through the helper. +package main + +import ( + "flag" + "fmt" + "go/ast" + "go/parser" + "go/token" + "io/fs" + "os" + "path/filepath" + "strings" +) + +// messageQueuePkg is the import path whose NewMessage is restricted. +const messageQueuePkg = "github.com/uber/submitqueue/platform/base/messagequeue" + +// allowedRoots are the directories permitted to construct messages directly. +// +// platform/publish owns the convention, so it is where the one remaining call +// belongs. The backends under platform/extension/messagequeue rebuild a Message +// from rows they are handed back on the read path, which is not a publish and +// chooses no ID. +var allowedRoots = []string{ + "platform/publish", + "platform/extension/messagequeue", +} + +// skipDirs are directory names never worth walking. +var skipDirs = map[string]bool{ + ".git": true, + "bazel-in": true, +} + +// violation is one direct construction of a queue message outside the +// allowed roots. +type violation struct { + file string + line int + fn string +} + +func main() { + flag.Parse() + + root, err := findRepoRoot() + if err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } + + var violations []violation + var checked int + + err = filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if d.IsDir() { + name := d.Name() + if skipDirs[name] || strings.HasPrefix(name, "bazel-") { + return fs.SkipDir + } + return nil + } + if !strings.HasSuffix(path, ".go") { + return nil + } + // Tests construct messages to feed a controller, which is the consuming + // side: nothing is published and no ID is chosen against a live topic. + if strings.HasSuffix(path, "_test.go") { + return nil + } + rel, relErr := filepath.Rel(root, path) + if relErr != nil { + rel = path + } + rel = filepath.ToSlash(rel) + if allowed(rel) { + return nil + } + checked++ + + found, checkErr := check(rel, path) + if checkErr != nil { + return checkErr + } + violations = append(violations, found...) + return nil + }) + if err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } + + if len(violations) > 0 { + fmt.Fprintf(os.Stderr, "%d direct queue-message construction(s) outside platform/publish:\n\n", len(violations)) + for _, v := range violations { + fmt.Fprintf(os.Stderr, " %s:%d: %s\n", v.file, v.line, v.fn) + } + fmt.Fprintf(os.Stderr, "\nA message ID is a deduplication key: a publish that reuses one is reported\n") + fmt.Fprintf(os.Stderr, "as a success and stores nothing, silently dropping the event. Publish through\n") + fmt.Fprintf(os.Stderr, "platform/publish and build the ID with publish.IntentID, naming the entity the\n") + fmt.Fprintf(os.Stderr, "message is about and the cause this particular message exists for.\n") + os.Exit(1) + } + + fmt.Printf("All %d files construct queue messages only through platform/publish.\n", checked) +} + +// allowed reports whether a repo-relative path may construct messages directly. +func allowed(rel string) bool { + for _, prefix := range allowedRoots { + if rel == prefix || strings.HasPrefix(rel, prefix+"/") { + return true + } + } + return false +} + +// check parses one file and reports every call constructing a queue message. +// +// Parsing rather than grepping, so a NewMessage of some unrelated package, or +// the name in a comment or string, is not mistaken for one of these. +func check(rel, path string) ([]violation, error) { + fset := token.NewFileSet() + file, err := parser.ParseFile(fset, path, nil, parser.SkipObjectResolution) + if err != nil { + return nil, fmt.Errorf("parse %s: %w", rel, err) + } + + // Names the file binds to the message-queue package. Usually one alias, + // but a file may import it more than once. + locals := map[string]bool{} + for _, imp := range file.Imports { + if strings.Trim(imp.Path.Value, `"`) != messageQueuePkg { + continue + } + switch { + case imp.Name == nil: + locals["messagequeue"] = true + case imp.Name.Name == "." || imp.Name.Name == "_": + // A dot import would make the call unqualified and a blank one + // cannot call anything; neither appears in this repo. + default: + locals[imp.Name.Name] = true + } + } + if len(locals) == 0 { + return nil, nil + } + + var violations []violation + ast.Inspect(file, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok { + return true + } + sel, ok := call.Fun.(*ast.SelectorExpr) + if !ok || !strings.HasPrefix(sel.Sel.Name, "NewMessage") { + return true + } + pkg, ok := sel.X.(*ast.Ident) + if !ok || !locals[pkg.Name] { + return true + } + violations = append(violations, violation{ + file: rel, + line: fset.Position(call.Pos()).Line, + fn: pkg.Name + "." + sel.Sel.Name, + }) + return true + }) + return violations, nil +} + +// findRepoRoot walks up from the working directory to the module root. +func findRepoRoot() (string, error) { + // Bazel `run` executes from the runfiles tree; BUILD_WORKSPACE_DIRECTORY + // points back at the source tree. + if dir := os.Getenv("BUILD_WORKSPACE_DIRECTORY"); dir != "" { + return dir, nil + } + dir, err := os.Getwd() + if err != nil { + return "", err + } + for { + if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil { + return dir, nil + } + parent := filepath.Dir(dir) + if parent == dir { + return "", fmt.Errorf("could not find repository root (no go.mod found)") + } + dir = parent + } +} diff --git a/tool/linter/messageid/main_test.go b/tool/linter/messageid/main_test.go new file mode 100644 index 00000000..9fcc6282 --- /dev/null +++ b/tool/linter/messageid/main_test.go @@ -0,0 +1,118 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCheck(t *testing.T) { + tests := []struct { + name string + src string + wantN int + wantF string + }{ + { + name: "aliased import is caught", + src: `package p +import entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" +func f() { _ = entityqueue.NewMessage("id", nil, "part", nil) }`, + wantN: 1, + wantF: "entityqueue.NewMessage", + }, + { + name: "unaliased import is caught under its package name", + src: `package p +import "github.com/uber/submitqueue/platform/base/messagequeue" +func f() { _ = messagequeue.NewMessage("id", nil, "part", nil) }`, + wantN: 1, + wantF: "messagequeue.NewMessage", + }, + { + name: "publishing through the helper passes", + src: `package p +import "github.com/uber/submitqueue/platform/publish" +func f() { _ = publish.IntentID("batch-1", "merged") }`, + }, + { + name: "NewMessage of an unrelated package passes", + src: `package p +import "example.com/other" +func f() { _ = other.NewMessage("id") }`, + }, + { + name: "the name in a comment or string passes", + src: `package p +// entityqueue.NewMessage is named here but not called. +func f() string { return "entityqueue.NewMessage(id, nil, part, nil)" }`, + }, + { + name: "importing without constructing passes", + src: `package p +import entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" +func f(m entityqueue.Message) string { return m.ID }`, + }, + { + name: "every construction in a file is reported", + src: `package p +import entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" +func f() { + _ = entityqueue.NewMessage("a", nil, "p", nil) + _ = entityqueue.NewMessage("b", nil, "p", nil) +}`, + wantN: 2, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "src.go") + require.NoError(t, os.WriteFile(path, []byte(tt.src), 0o600)) + + got, err := check("src.go", path) + require.NoError(t, err) + require.Len(t, got, tt.wantN) + if tt.wantF != "" { + assert.Equal(t, tt.wantF, got[0].fn) + } + }) + } +} + +func TestAllowed(t *testing.T) { + tests := []struct { + path string + want bool + }{ + {"platform/publish/publish.go", true}, + {"platform/extension/messagequeue/mysql/subscriber.go", true}, + {"submitqueue/orchestrator/controller/batch/batch.go", false}, + {"runway/controller/merge/merge.go", false}, + // A path merely prefixed by an allowed root's name is not inside it. + {"platform/publishing/thing.go", false}, + } + + for _, tt := range tests { + t.Run(tt.path, func(t *testing.T) { + assert.Equal(t, tt.want, allowed(tt.path)) + }) + } +}