From a0106c0ba9814ba8cc75663f525c7714185cee2d Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Tue, 11 Aug 2026 20:00:47 -0700 Subject: [PATCH] fix(messagequeue): intent-scoped message IDs so wake-ups stop vanishing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The MySQL queue deduplicates publishes on `(topic, partition_key, id)` via `INSERT ... ON DUPLICATE KEY UPDATE topic = topic`, and rows are removed only by `GarbageCollect` — which runs from `subscriber.go` on idle ticks only, with `gcCounter` reset to `0` by any tick that delivered a message. On a busy partition GC never runs, so the dedup horizon is unbounded exactly when traffic is high. A publish that collides is reported as a success, writes nothing, and has no error to retry and no row to deliver. Controllers reusing a bare entity ID as the message ID therefore lose their *second, unrelated* publish about that entity. Concretely: `batch` announces a new batch to speculate under the batch ID; when that batch later merges, `mergesignal.fanout`'s "wake the dependents" publish reuses the same ID and is dropped against the announcement. `fanout` returns nil and the delivery is acked. Other speculate publishers already mint distinct IDs, so another batch's build signal usually re-plans the queue and hides this — the stall shows at the tail, when the merged batch is the last in flight and nothing else pings speculate. Fixing only that call site would leave the shape in place. `submitqueue/core/publish` documented the hazard but was domain-scoped, so `runway/` and `stovepipe/` published bare entity IDs with no shared guidance, and `platform/base/messagequeue` documented none of it. ### What? `submitqueue/core/publish` moves to `platform/publish` — it already imported only `platform/base/messagequeue` and `platform/consumer`, so this is a relocation, and it lets every domain share one helper instead of hand-rolling the resolve-registry-and-publish block three more times. The new `publish.IntentID(entityID, cause...)` names the occasion to publish rather than the entity published about. A retry of the same cause dedups, which is what keeps redelivery safe; a new cause about the same entity can never be swallowed. Deterministic IDs go where a duplicate is harmful and the cause is nameable: | Publish | ID | | --- | --- | | merge result → speculate | `{batch}/merged` (**the bug**) | | merge result → conclude | `{batch}/conclude/merged` | | build poll → speculate | `{batch}/build-signal/{build}/{status}` | | speculate → merge dispatch | `{batch}/merge-dispatch` | | speculate → conclude on terminal | `{batch}/conclude/speculate` | | one-shot hand-offs | bare `IntentID(id)` | `buildsignal` publishes to speculate on *every* poll, so keying on `{batch}/build-signal/{build}` would have deduplicated the terminal wake-up into the first poll's. Including the observed status lands every transition and collapses only the polls that saw nothing new — a reduction in speculate churn versus the previous `UniqueID`. `publish.UniqueID` is kept, and deliberately left in place, for repeat-until-effective nudges whose provoking condition is that nothing recorded the last one — speculate's build dispatch, its self-heal fan-out and `recoverable`, cancel's nudge, the DLQ re-trigger. Those have no stable cause to name, and a deterministic ID would dedup the re-send against the message that went missing. Each now says so at the call site. `runway/controller/dlq` gets its own cause: it answered on the same topic under the same correlation ID as the live handler, so a dead-lettered request could have its terminal failure deduplicated against an answer already sent. `tool/linter/messageid` makes the helper the only door. Judging whether an ID expression is well chosen is not decidable by reading it, so the linter enforces the structural rule instead: `NewMessage` may only be called from `platform/publish` and the queue backends. It found seven production call sites the manual audit missed. ## Test Plan - ✅ `bazel test //...` — 99 unit tests - ✅ `bazel test //test/integration/...` — 8 suites - ✅ `bazel test //test/e2e/...` — 3 suites - ✅ `make lint-license`, `make lint-message-id`, `make lint-queue-shard`; `make fmt` idempotent; `make tidy` clean - New `TestProcess_FanoutDoesNotCollideWithTheBatchAnnouncement` was confirmed to fail against the old code — reverting the one expression reproduces the drop. - New `TestDedupOutlivesConsumption` pins the underlying behaviour against real MySQL: a consumed and acked message still deduplicates a later publish under the same ID, and naming the cause gets it through. - Not covered: no e2e exercises a dependent chain, so the merge→dependent-wake path is not verified end to end. Building that fixture is follow-up work. Out of scope, deliberately: GC never running on a busy partition. It widens this window and also lets `queue_messages` grow without bound, but it is orthogonal to the ID convention and wants its own review. ## Issue Closes #352 --- Makefile | 5 +- platform/base/messagequeue/message.go | 11 + platform/extension/messagequeue/README.md | 11 + .../core => platform}/publish/BUILD.bazel | 2 +- platform/publish/publish.go | 102 ++++++++ .../core => platform}/publish/publish_test.go | 41 ++++ runway/controller/dlq/BUILD.bazel | 2 +- runway/controller/dlq/dlq.go | 22 +- runway/controller/merge/BUILD.bazel | 2 +- runway/controller/merge/merge.go | 22 +- .../controller/mergeconflictcheck/BUILD.bazel | 2 +- .../mergeconflictcheck/mergeconflictcheck.go | 20 +- stovepipe/controller/BUILD.bazel | 2 +- stovepipe/controller/build/BUILD.bazel | 2 +- stovepipe/controller/build/build.go | 17 +- stovepipe/controller/buildsignal/BUILD.bazel | 2 +- .../controller/buildsignal/buildsignal.go | 24 +- stovepipe/controller/ingest.go | 18 +- stovepipe/controller/process/BUILD.bazel | 2 +- stovepipe/controller/process/process.go | 15 +- submitqueue/core/publish/publish.go | 67 ----- submitqueue/core/request/BUILD.bazel | 2 +- submitqueue/core/request/log.go | 20 +- submitqueue/gateway/controller/BUILD.bazel | 2 +- submitqueue/gateway/controller/cancel.go | 22 +- submitqueue/gateway/controller/land.go | 22 +- .../orchestrator/controller/batch/BUILD.bazel | 2 +- .../orchestrator/controller/batch/batch.go | 24 +- .../orchestrator/controller/build/BUILD.bazel | 2 +- .../orchestrator/controller/build/build.go | 11 +- .../controller/buildsignal/BUILD.bazel | 2 +- .../controller/buildsignal/buildsignal.go | 21 +- .../buildsignal/buildsignal_test.go | 42 ++++ .../controller/cancel/BUILD.bazel | 2 +- .../orchestrator/controller/cancel/cancel.go | 2 +- .../orchestrator/controller/dlq/BUILD.bazel | 2 +- .../orchestrator/controller/dlq/speculate.go | 2 +- .../orchestrator/controller/merge/BUILD.bazel | 2 +- .../orchestrator/controller/merge/merge.go | 20 +- .../mergeconflictsignal/BUILD.bazel | 2 +- .../mergeconflictsignal.go | 20 +- .../controller/mergesignal/BUILD.bazel | 2 +- .../controller/mergesignal/mergesignal.go | 34 ++- .../mergesignal/mergesignal_test.go | 53 ++++ .../controller/speculate/BUILD.bazel | 2 +- .../controller/speculate/dispatch.go | 8 +- .../controller/speculate/finalize.go | 20 +- .../controller/speculate/run_test.go | 15 +- .../controller/speculate/speculate.go | 29 ++- .../orchestrator/controller/start/BUILD.bazel | 2 +- .../orchestrator/controller/start/start.go | 19 +- .../controller/validate/BUILD.bazel | 2 +- .../controller/validate/validate.go | 20 +- .../messagequeue/mysql/queue_test.go | 52 ++++ tool/linter/messageid/BUILD.bazel | 24 ++ tool/linter/messageid/main.go | 228 ++++++++++++++++++ tool/linter/messageid/main_test.go | 118 +++++++++ 57 files changed, 886 insertions(+), 357 deletions(-) rename {submitqueue/core => platform}/publish/BUILD.bazel (91%) create mode 100644 platform/publish/publish.go rename {submitqueue/core => platform}/publish/publish_test.go (68%) delete mode 100644 submitqueue/core/publish/publish.go create mode 100644 tool/linter/messageid/BUILD.bazel create mode 100644 tool/linter/messageid/main.go create mode 100644 tool/linter/messageid/main_test.go 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)) + }) + } +}