fix(messagequeue): intent-scoped message IDs so wake-ups stop vanishing

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
 

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.

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.

submitqueue/core/publish/BUILD.bazel → 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",

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))
+}

submitqueue/core/publish/publish_test.go → 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")

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",
     ],

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)
 	}
 

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",

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)
 	}
 

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",

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)
 	}
 

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",