feat(submitqueue): cut over internal queues to protojson

AGENTS.md

@@ -168,7 +168,7 @@ Paths follow the directory layout: shared packages live under `platform/` at the
 - Queue contracts: external `github.com/uber/submitqueue/api/{domain}/messagequeue`; internal `github.com/uber/submitqueue/{domain}/core/messagequeue`
 - Domain entities: `github.com/uber/submitqueue/{domain}/entity` (e.g. `.../submitqueue/entity`)
 - Domain extensions: `github.com/uber/submitqueue/{domain}/extension/{ext}[/{impl}]` (e.g. `.../submitqueue/extension/storage/mysql`)
-- Cross-domain consumer framework: `github.com/uber/submitqueue/platform/consumer`; internal topic keys live with the owning domain contract (for example `submitqueue/core/topickey` and `stovepipe/core/messagequeue`); external queue topic keys live with their published contract (for example `api/runway/messagequeue`)
+- Cross-domain consumer framework: `github.com/uber/submitqueue/platform/consumer`; internal topic keys live with the owning domain contract (for example `submitqueue/core/messagequeue` and `stovepipe/core/messagequeue`); external queue topic keys live with their published contract (for example `api/runway/messagequeue`)
 - Domain-internal infra: `github.com/uber/submitqueue/{domain}/core/{pkg}` (e.g. `.../submitqueue/core/request`)
 - Shared entities: `github.com/uber/submitqueue/platform/base/{pkg}` (e.g. `.../platform/base/messagequeue`)
 - Shared extensions: `github.com/uber/submitqueue/platform/extension/{ext}[/{impl}]` (e.g. `.../platform/extension/messagequeue/mysql`)
@@ -200,7 +200,7 @@ New queue contracts are defined in **proto3** (`.proto` under `proto/`, generate
 
 The message types are generated; the contract package adds only generic `protojson` glue — `Marshal(m)` / `Unmarshal[T](b, m)` — owning the wire conventions: `UseProtoNames` (snake_case fields), UPPER_SNAKE enum values, int64-as-string, unknown fields discarded on read (additive evolution). The topic key(s) carrying a message are declared on the message via the `topic_keys` proto option — a `google.protobuf.MessageOptions` extension defined in `api/base/messagequeue`. A topic key is a stable logical name, not a concrete wire topic; each implementer maps it to its backend's topic name, and a `TopicKeys(msg)` reflection helper reads the option back. It is contract metadata, not the hot path — publish/consume still routes on `consumer.TopicKey` + `TopicRegistry`. The contract package owns both halves: the proto payload and the `TopicKey` constants for its topic keys. A contract test round-trips the payloads and asserts every topic key is bound to exactly one message. Shared field types (`Change`, `Strategy`) are shared protos under `api/base/{change,mergestrategy}`. `api/runway/messagequeue/` is the reference example.
 
-SubmitQueue's internal pipeline predates the proto-backed convention. It continues to serialize domain entities with `encoding/json` and declares its logical keys in `submitqueue/core/topickey/`. Do not convert or mix these wire formats incidentally; treat migration as an explicit compatibility change.
+SubmitQueue's internal pipeline contract lives at `submitqueue/core/messagequeue/` (proto3 + protojson, one message per topic key). `MarshalID`/`UnmarshalID` take the topic key so consume uses the bound message. Proto filenames that would collide with another domain in the protobuf registry are prefixed (`submitqueuemerge.proto`, `submitqueuebuild.proto`, `submitqueuebuildsignal.proto`). `submitqueue/core/topickey` re-exports the topic-key constants from that package.
 
 ### Naming Conventions
 

Makefile

@@ -36,7 +36,7 @@ GOIMPORTS_VERSION ?= v0.33.0
 # (the out_dir convention in tool/proto/BUILD.bazel) and copied back here. A
 # package may hold multiple .proto files (e.g. an RPC contract plus messagequeue
 # contracts); all generated stubs land in the same protopb/ dir.
-PROTO_PACKAGES = api/base/change api/base/hook api/base/mergestrategy api/base/messagequeue api/runway/messagequeue api/runway api/submitqueue/gateway api/submitqueue/orchestrator api/stovepipe stovepipe/core/messagequeue
+PROTO_PACKAGES = api/base/change api/base/hook api/base/mergestrategy api/base/messagequeue api/runway/messagequeue api/runway api/submitqueue/gateway api/submitqueue/orchestrator api/stovepipe stovepipe/core/messagequeue submitqueue/core/messagequeue
 
 # Set REPO_ROOT for docker-compose
 export REPO_ROOT := $(shell pwd)

submitqueue/core/messagequeue/BUILD.bazel

@@ -0,0 +1,43 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "id.go",
+        "map.go",
+        "messagequeue.go",
+        "topics.go",
+    ],
+    importpath = "github.com/uber/submitqueue/submitqueue/core/messagequeue",
+    visibility = [
+        "//service/submitqueue:__subpackages__",
+        "//submitqueue:__subpackages__",
+        "//test:__subpackages__",
+    ],
+    deps = [
+        "//api/base/change/protopb:go_default_library",
+        "//api/base/mergestrategy/protopb:go_default_library",
+        "//api/base/messagequeue/protopb:go_default_library",
+        "//platform/base/change:go_default_library",
+        "//platform/base/mergestrategy:go_default_library",
+        "//platform/consumer:go_default_library",
+        "//submitqueue/core/messagequeue/protopb:go_default_library",
+        "//submitqueue/entity:go_default_library",
+        "@org_golang_google_protobuf//encoding/protojson:go_default_library",
+        "@org_golang_google_protobuf//proto:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["messagequeue_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/base/change:go_default_library",
+        "//platform/base/mergestrategy:go_default_library",
+        "//submitqueue/entity:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@org_golang_google_protobuf//proto:go_default_library",
+    ],
+)

submitqueue/core/messagequeue/README.md

@@ -0,0 +1,25 @@
+# SubmitQueue internal message-queue contract
+
+Wire payloads for the queues internal to the SubmitQueue pipeline (gateway and orchestrator). It is **internal** — used only within the SubmitQueue domain — so it lives under `submitqueue/core` rather than `api/` (Bazel visibility keeps it domain-scoped).
+
+Payloads are defined in proto3 (`proto/`, generated into `protopb/`) and serialized as **protobuf JSON** (protojson), so the MySQL-backed queue keeps storing self-describing JSON. The contract package adds protojson glue (`Marshal`/`Unmarshal`), `TopicKeys`, the pipeline `TopicKey` constants, and helpers that map generated payloads to `submitqueue/entity` types. `MarshalID`/`UnmarshalID` take the topic key so the bound message is used; unmarshalling through a different type would drop fields added later (`DiscardUnknown`). Each payload declares the topic key that carries it via the `topic_keys` proto option (defined in `api/base/messagequeue`); a contract test round-trips every payload and asserts each topic key is bound to exactly one message.
+
+Shared field types `Change` and `Strategy` come from `api/base/change` and `api/base/mergestrategy`.
+
+## Stages
+
+Each topic key has its own message, even when the first version is only an id and a queue, so a stage can grow fields without touching others.
+
+- **start** (`TopicKeyStart`, `Start`) — gateway publishes the minted request id and land inputs; start persists a `Request`. Full payload: this seam crosses services.
+- **cancel** (`TopicKeyCancel`, `Cancel`) — gateway publishes the request id to cancel; cancel reloads the `Request`. Full payload across the gateway/orchestrator seam.
+- **validate** (`TopicKeyValidate`, `Validate`) — start publishes the request id; validate reloads the `Request`.
+- **batch** (`TopicKeyBatch`, `Batch`) — landconflictsignal publishes the request id; batch reloads the `Request`.
+- **dependency-analysis** (`TopicKeyDependencyAnalysis`, `DependencyAnalysis`) — batch publishes the batch id; partitioned by queue.
+- **speculate** (`TopicKeySpeculate`, `Speculate`) — dependency-analysis (and later stages) publish the batch id.
+- **build** (`TopicKeyBuild`, `Build` in `submitqueuebuild.proto`) — speculate publishes a **batch** id; build reloads the `Batch`. The proto filename is not `build.proto` so it does not collide with Stovepipe's `stovepipe/core/messagequeue/proto/build.proto` in the protobuf filename registry.
+- **buildsignal** (`TopicKeyBuildSignal`, `BuildSignal` in `submitqueuebuildsignal.proto`) — build publishes a **build** id; buildsignal polls and may hold the delivery. Same filename-registry reason as build.
+- **submitqueue-land** (`TopicKeyLand`, `Merge` in `submitqueuemerge.proto`) — speculate publishes a batch id; land reloads the `Batch` before handing work to Runway. The proto filename is not `merge.proto`/`land.proto` so it does not collide with Runway's `api/runway/messagequeue/proto/merge.proto` in the protobuf filename registry.
+- **conclude** (`TopicKeyConclude`, `Conclude`) — speculate/landsignal publish a batch id. A failed batch's reason travels in message metadata (`MetadataKeyFailureReason`), not the payload.
+- **log** (`TopicKeyLog`, `Log`) — orchestrator publishes a full request-log entry; the gateway materializes it. `type`, `status`, and `event` are open strings matching the domain vocabularies.
+
+In-boundary stages (validate through conclude, except start/cancel/log) put only an id on the queue because producer and consumer share storage.

submitqueue/core/messagequeue/id.go

@@ -0,0 +1,143 @@
+// 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 messagequeue
+
+import (
+	"fmt"
+
+	"google.golang.org/protobuf/proto"
+
+	"github.com/uber/submitqueue/submitqueue/entity"
+)
+
+type idPayload interface {
+	proto.Message
+	GetId() string
+	GetQueue() string
+}
+
+func idOnlyMessage(key TopicKey, id, queue string) (idPayload, error) {
+	switch key {
+	case TopicKeyValidate:
+		return &Validate{Id: id, Queue: queue}, nil
+	case TopicKeyBatch:
+		return &Batch{Id: id, Queue: queue}, nil
+	case TopicKeyDependencyAnalysis:
+		return &DependencyAnalysis{Id: id, Queue: queue}, nil
+	case TopicKeySpeculate:
+		return &Speculate{Id: id, Queue: queue}, nil
+	case TopicKeyBuild:
+		return &Build{Id: id, Queue: queue}, nil
+	case TopicKeyBuildSignal:
+		return &BuildSignal{Id: id, Queue: queue}, nil
+	case TopicKeyLand:
+		return &Merge{Id: id, Queue: queue}, nil
+	case TopicKeyConclude:
+		return &Conclude{Id: id, Queue: queue}, nil
+	default:
+		return nil, fmt.Errorf("topic %q does not carry an id-only payload", key)
+	}
+}
+
+// MarshalID serializes the id-only payload bound to key. Start, cancel, and log
+// are not id-only; callers marshal those messages directly.
+func MarshalID(key TopicKey, id, queue string) ([]byte, error) {
+	m, err := idOnlyMessage(key, id, queue)
+	if err != nil {
+		return nil, err
+	}
+	return Marshal(m)
+}
+
+// UnmarshalID reads id and queue from the id-only payload bound to key.
+// Consumers pass the topic they subscribe to so a field added to that message
+// is decoded rather than discarded as unknown on a different type.
+func UnmarshalID(key TopicKey, b []byte) (id, queue string, err error) {
+	m, err := idOnlyMessage(key, "", "")
+	if err != nil {
+		return "", "", err
+	}
+	if err := Unmarshal(b, m); err != nil {
+		return "", "", err
+	}
+	return m.GetId(), m.GetQueue(), nil
+}
+
+// UnmarshalLandRequest reads a start payload into the gateway-owned land request.
+func UnmarshalLandRequest(b []byte) (entity.LandRequest, error) {
+	m := &Start{}
+	if err := Unmarshal(b, m); err != nil {
+		return entity.LandRequest{}, err
+	}
+	return LandRequestFromStart(m), nil
+}
+
+// UnmarshalCancelRequest reads a cancel payload into the domain cancellation.
+func UnmarshalCancelRequest(b []byte) (entity.CancelRequest, error) {
+	m := &Cancel{}
+	if err := Unmarshal(b, m); err != nil {
+		return entity.CancelRequest{}, err
+	}
+	return CancelToEntity(m), nil
+}
+
+// UnmarshalRequestLog reads a log payload into a request-log entry.
+func UnmarshalRequestLog(b []byte) (entity.RequestLog, error) {
+	m := &Log{}
+	if err := Unmarshal(b, m); err != nil {
+		return entity.RequestLog{}, err
+	}
+	return LogToEntity(m), nil
+}
+
+// UnmarshalRequestID reads a request-scoped id-only payload (validate or batch).
+func UnmarshalRequestID(key TopicKey, b []byte) (entity.RequestID, error) {
+	switch key {
+	case TopicKeyValidate, TopicKeyBatch:
+	default:
+		return entity.RequestID{}, fmt.Errorf("topic %q does not carry a request-id payload", key)
+	}
+	id, queue, err := UnmarshalID(key, b)
+	if err != nil {
+		return entity.RequestID{}, err
+	}
+	return entity.RequestID{ID: id, Queue: queue}, nil
+}
+
+// UnmarshalBatchID reads a batch-scoped id-only payload.
+func UnmarshalBatchID(key TopicKey, b []byte) (entity.BatchID, error) {
+	switch key {
+	case TopicKeyDependencyAnalysis, TopicKeySpeculate, TopicKeyBuild, TopicKeyLand, TopicKeyConclude:
+	default:
+		return entity.BatchID{}, fmt.Errorf("topic %q does not carry a batch-id payload", key)
+	}
+	id, queue, err := UnmarshalID(key, b)
+	if err != nil {
+		return entity.BatchID{}, err
+	}
+	return entity.BatchID{ID: id, Queue: queue}, nil
+}
+
+// UnmarshalBuildID reads a buildsignal payload.
+func UnmarshalBuildID(key TopicKey, b []byte) (entity.BuildID, error) {
+	if key != TopicKeyBuildSignal {
+		return entity.BuildID{}, fmt.Errorf("topic %q does not carry a build-id payload", key)
+	}
+	id, queue, err := UnmarshalID(key, b)
+	if err != nil {
+		return entity.BuildID{}, err
+	}
+	return entity.BuildID{ID: id, Queue: queue}, nil
+}

submitqueue/core/messagequeue/map.go

@@ -0,0 +1,127 @@
+// 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 messagequeue
+
+import (
+	changepb "github.com/uber/submitqueue/api/base/change/protopb"
+	strategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb"
+	"github.com/uber/submitqueue/platform/base/change"
+	"github.com/uber/submitqueue/platform/base/mergestrategy"
+	"github.com/uber/submitqueue/submitqueue/entity"
+)
+
+// StartFromLandRequest copies a gateway-owned land request onto the start payload.
+func StartFromLandRequest(r entity.LandRequest) *Start {
+	return &Start{
+		Id:           r.ID,
+		Queue:        r.Queue,
+		Change:       &changepb.Change{Uris: append([]string{}, r.Change.URIs...)},
+		LandStrategy: landStrategyToProto(r.LandStrategy),
+	}
+}
+
+// LandRequestFromStart copies a start payload onto the gateway-owned land request.
+func LandRequestFromStart(m *Start) entity.LandRequest {
+	var uris []string
+	if m.GetChange() != nil {
+		uris = append([]string{}, m.GetChange().GetUris()...)
+	}
+	return entity.LandRequest{
+		ID:           m.GetId(),
+		Queue:        m.GetQueue(),
+		Change:       change.Change{URIs: uris},
+		LandStrategy: landStrategyFromProto(m.GetLandStrategy()),
+	}
+}
+
+// CancelFromEntity copies a cancellation onto the cancel payload.
+func CancelFromEntity(r entity.CancelRequest) *Cancel {
+	return &Cancel{Id: r.ID, Queue: r.Queue, Reason: r.Reason}
+}
+
+// CancelToEntity copies a cancel payload onto the domain cancellation.
+func CancelToEntity(m *Cancel) entity.CancelRequest {
+	return entity.CancelRequest{ID: m.GetId(), Queue: m.GetQueue(), Reason: m.GetReason()}
+}
+
+// LogFromEntity copies a request-log entry onto the log payload.
+func LogFromEntity(r entity.RequestLog) *Log {
+	return &Log{
+		RequestId:      r.RequestID,
+		Queue:          r.Queue,
+		TimestampMs:    r.TimestampMs,
+		Type:           string(r.Type),
+		Status:         string(r.Status),
+		Event:          string(r.Event),
+		RequestVersion: r.RequestVersion,
+		LastError:      r.LastError,
+		Metadata:       r.Metadata,
+	}
+}
+
+// LogToEntity copies a log payload onto a request-log entry. An empty type is
+// treated as a status entry, matching entries written before the type field
+// existed. A nil metadata map becomes an empty map.
+func LogToEntity(m *Log) entity.RequestLog {
+	meta := m.GetMetadata()
+	if meta == nil {
+		meta = make(map[string]string)
+	}
+	typ := entity.RequestLogType(m.GetType())
+	if typ == "" {
+		typ = entity.RequestLogTypeStatus
+	}
+	return entity.RequestLog{
+		RequestID:      m.GetRequestId(),
+		Queue:          m.GetQueue(),
+		TimestampMs:    m.GetTimestampMs(),
+		Type:           typ,
+		Status:         entity.RequestStatus(m.GetStatus()),
+		Event:          entity.RequestEvent(m.GetEvent()),
+		RequestVersion: m.GetRequestVersion(),
+		LastError:      m.GetLastError(),
+		Metadata:       meta,
+	}
+}
+
+func landStrategyToProto(s mergestrategy.MergeStrategy) strategypb.Strategy {
+	switch s {
+	case mergestrategy.MergeStrategyRebase:
+		return strategypb.Strategy_REBASE
+	case mergestrategy.MergeStrategySquashRebase:
+		return strategypb.Strategy_SQUASH_REBASE
+	case mergestrategy.MergeStrategyMerge:
+		return strategypb.Strategy_MERGE
+	case mergestrategy.MergeStrategyPromote:
+		return strategypb.Strategy_PROMOTE
+	default:
+		return strategypb.Strategy_DEFAULT
+	}
+}
+
+func landStrategyFromProto(s strategypb.Strategy) mergestrategy.MergeStrategy {
+	switch s {
+	case strategypb.Strategy_REBASE:
+		return mergestrategy.MergeStrategyRebase
+	case strategypb.Strategy_SQUASH_REBASE:
+		return mergestrategy.MergeStrategySquashRebase
+	case strategypb.Strategy_MERGE:
+		return mergestrategy.MergeStrategyMerge
+	case strategypb.Strategy_PROMOTE:
+		return mergestrategy.MergeStrategyPromote
+	default:
+		return mergestrategy.MergeStrategyUnknown
+	}
+}