diff --git a/stovepipe/core/messagequeue/BUILD.bazel b/stovepipe/core/messagequeue/BUILD.bazel index 4920c8eb..f32ddbd0 100644 --- a/stovepipe/core/messagequeue/BUILD.bazel +++ b/stovepipe/core/messagequeue/BUILD.bazel @@ -19,7 +19,7 @@ go_library( go_test( name = "go_default_test", - srcs = ["process_test.go"], + srcs = ["messagequeue_test.go"], embed = [":go_default_library"], deps = [ "@com_github_stretchr_testify//assert:go_default_library", diff --git a/stovepipe/core/messagequeue/README.md b/stovepipe/core/messagequeue/README.md index 2a4b32a3..15889fe1 100644 --- a/stovepipe/core/messagequeue/README.md +++ b/stovepipe/core/messagequeue/README.md @@ -7,5 +7,7 @@ Payloads are defined in proto3 (`proto/`, generated into `protopb/`) and seriali ## Stages - **process** (`TopicKeyProcess`, `ProcessRequest`) — ingest publishes the minted request id here once it accepts a new head; the process controller reloads the `Request` from storage and decides the build strategy. Only the id travels: producer and consumer share the store, so messages stay small and redelivery is idempotent. +- **build** (`TopicKeyBuild`, `BuildRequest`) — process/analyze publishes the request id here once its build scope (`BuildStrategy`/`BaseURI`) is decided; the build controller reloads the `Request`, triggers the build, and persists the resulting `Build`. Partitioned by request id. +- **buildsignal** (`TopicKeyBuildSignal`, `BuildSignal`) — build publishes the build id here after triggering; buildsignal re-publishes to itself between polls until the build reaches a terminal status. Partitioned by build id, so each build's poll loop is an independent partition. See [doc/rfc/stovepipe/steps/build.md](../../../doc/rfc/stovepipe/steps/build.md) and [buildsignal.md](../../../doc/rfc/stovepipe/steps/buildsignal.md). See [doc/rfc/messagequeue-contract.md](../../../doc/rfc/messagequeue-contract.md) for the contract conventions and `api/runway/messagequeue` for the external reference example. diff --git a/stovepipe/core/messagequeue/messagequeue.go b/stovepipe/core/messagequeue/messagequeue.go index e88933e1..0beb8f2a 100644 --- a/stovepipe/core/messagequeue/messagequeue.go +++ b/stovepipe/core/messagequeue/messagequeue.go @@ -42,9 +42,13 @@ type ( // minted request id to validate. ProcessRequest = protopb.ProcessRequest - // BuildRequest is the payload process publishes to the build stage: the - // admitted request id whose persisted strategy and baseline build reloads. + // BuildRequest is the payload process/analyze publishes to the build stage: + // the request id to build. BuildRequest = protopb.BuildRequest + + // BuildSignal is the payload build publishes to the buildsignal stage, and + // buildsignal re-publishes to itself while polling: the build id to poll. + BuildSignal = protopb.BuildSignal ) // marshalOpts keeps the JSON field names identical to the proto field names diff --git a/stovepipe/core/messagequeue/process_test.go b/stovepipe/core/messagequeue/messagequeue_test.go similarity index 88% rename from stovepipe/core/messagequeue/process_test.go rename to stovepipe/core/messagequeue/messagequeue_test.go index 185de8de..9d1635db 100644 --- a/stovepipe/core/messagequeue/process_test.go +++ b/stovepipe/core/messagequeue/messagequeue_test.go @@ -44,6 +44,17 @@ func TestBuildRequestRoundTrip(t *testing.T) { assert.True(t, proto.Equal(req, got), "round-tripped BuildRequest should equal the original") } +func TestBuildSignalRoundTrip(t *testing.T) { + sig := &BuildSignal{Id: "bk-1001"} + + data, err := Marshal(sig) + require.NoError(t, err) + + got := &BuildSignal{} + require.NoError(t, Unmarshal(data, got)) + assert.True(t, proto.Equal(sig, got), "round-tripped BuildSignal should equal the original") +} + // TestWireFormat locks the protojson encoding decision the contract relies on: // snake_case field names (UseProtoNames). func TestWireFormat(t *testing.T) { @@ -58,7 +69,7 @@ func TestWireFormat(t *testing.T) { // no topic_keys option names an unknown key. func TestTopicKeysBindEveryTopicKey(t *testing.T) { bound := map[string]int{} - for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}} { + for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}} { keys := TopicKeys(m) require.NotEmpty(t, keys, "message must declare a non-empty topic_keys option") for _, key := range keys { @@ -69,6 +80,7 @@ func TestTopicKeysBindEveryTopicKey(t *testing.T) { keys := []TopicKey{ TopicKeyProcess, TopicKeyBuild, + TopicKeyBuildSignal, } valid := map[string]bool{} diff --git a/stovepipe/core/messagequeue/proto/BUILD.bazel b/stovepipe/core/messagequeue/proto/BUILD.bazel index 95634720..f625ba47 100644 --- a/stovepipe/core/messagequeue/proto/BUILD.bazel +++ b/stovepipe/core/messagequeue/proto/BUILD.bazel @@ -1,6 +1,7 @@ exports_files( [ "build.proto", + "buildsignal.proto", "process.proto", ], visibility = ["//tool/proto:__pkg__"], diff --git a/stovepipe/core/messagequeue/proto/build.proto b/stovepipe/core/messagequeue/proto/build.proto index 5b607614..b30ce26e 100644 --- a/stovepipe/core/messagequeue/proto/build.proto +++ b/stovepipe/core/messagequeue/proto/build.proto @@ -23,12 +23,14 @@ option java_multiple_files = true; option java_outer_classname = "BuildProto"; option java_package = "com.uber.submitqueue.stovepipe.messagequeue"; -// BuildRequest is the payload process publishes after admitting a request. Only -// the request id travels; build reloads the request from storage and uses the -// strategy and baseline process persisted at admission. +// BuildRequest is the payload process/analyze publishes to the build stage once +// the request's build scope (BuildStrategy/BaseURI) has been decided: only the +// request id travels on the queue. build reloads the full Request from storage +// by this id (producer and consumer share the store, so the id is enough and +// redelivery stays idempotent). See doc/rfc/stovepipe/steps/build.md. message BuildRequest { option (uber.base.messagequeue.topic_keys) = "build"; - // id is the admitted request id to build. Format: "request//". + // id is the request id to build. Format: "request//". string id = 1; } diff --git a/stovepipe/core/messagequeue/proto/buildsignal.proto b/stovepipe/core/messagequeue/proto/buildsignal.proto new file mode 100644 index 00000000..3868ad9e --- /dev/null +++ b/stovepipe/core/messagequeue/proto/buildsignal.proto @@ -0,0 +1,37 @@ +// 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. + +syntax = "proto3"; + +package uber.stovepipe.messagequeue; + +import "api/base/messagequeue/proto/messagequeue.proto"; + +option go_package = "github.com/uber/submitqueue/stovepipe/core/messagequeue/protopb"; +option java_multiple_files = true; +option java_outer_classname = "BuildSignalProto"; +option java_package = "com.uber.submitqueue.stovepipe.messagequeue"; + +// BuildSignal is the payload build publishes to the buildsignal stage, and +// buildsignal re-publishes to itself while polling: only the build id travels +// on the queue. buildsignal reloads the full Build from storage by this id +// (producer and consumer share the store, so the id is enough and redelivery +// stays idempotent). See doc/rfc/stovepipe/steps/buildsignal.md. +message BuildSignal { + option (uber.base.messagequeue.topic_keys) = "buildsignal"; + + // id is the build id to poll: the runner-assigned id minted by + // BuildRunner.Trigger (entity.Build.ID / entity.BuildID.ID). + string id = 1; +} diff --git a/stovepipe/core/messagequeue/protopb/BUILD.bazel b/stovepipe/core/messagequeue/protopb/BUILD.bazel index 9a16d825..5c59a722 100644 --- a/stovepipe/core/messagequeue/protopb/BUILD.bazel +++ b/stovepipe/core/messagequeue/protopb/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "go_default_library", srcs = [ "build.pb.go", + "buildsignal.pb.go", "process.pb.go", ], importpath = "github.com/uber/submitqueue/stovepipe/core/messagequeue/protopb", diff --git a/stovepipe/core/messagequeue/protopb/build.pb.go b/stovepipe/core/messagequeue/protopb/build.pb.go index 60025cf8..e8150153 100644 --- a/stovepipe/core/messagequeue/protopb/build.pb.go +++ b/stovepipe/core/messagequeue/protopb/build.pb.go @@ -37,12 +37,14 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) -// BuildRequest is the payload process publishes after admitting a request. Only -// the request id travels; build reloads the request from storage and uses the -// strategy and baseline process persisted at admission. +// BuildRequest is the payload process/analyze publishes to the build stage once +// the request's build scope (BuildStrategy/BaseURI) has been decided: only the +// request id travels on the queue. build reloads the full Request from storage +// by this id (producer and consumer share the store, so the id is enough and +// redelivery stays idempotent). See doc/rfc/stovepipe/steps/build.md. type BuildRequest struct { state protoimpl.MessageState `protogen:"open.v1"` - // id is the admitted request id to build. Format: "request//". + // id is the request id to build. Format: "request//". Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache diff --git a/stovepipe/core/messagequeue/protopb/buildsignal.pb.go b/stovepipe/core/messagequeue/protopb/buildsignal.pb.go new file mode 100644 index 00000000..2e4eff18 --- /dev/null +++ b/stovepipe/core/messagequeue/protopb/buildsignal.pb.go @@ -0,0 +1,146 @@ +// 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. + +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.10 +// protoc v5.29.3 +// source: buildsignal.proto + +package protopb + +import ( + reflect "reflect" + sync "sync" + unsafe "unsafe" + + _ "github.com/uber/submitqueue/api/base/messagequeue/protopb" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// BuildSignal is the payload build publishes to the buildsignal stage, and +// buildsignal re-publishes to itself while polling: only the build id travels +// on the queue. buildsignal reloads the full Build from storage by this id +// (producer and consumer share the store, so the id is enough and redelivery +// stays idempotent). See doc/rfc/stovepipe/steps/buildsignal.md. +type BuildSignal struct { + state protoimpl.MessageState `protogen:"open.v1"` + // id is the build id to poll: the runner-assigned id minted by + // BuildRunner.Trigger (entity.Build.ID / entity.BuildID.ID). + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *BuildSignal) Reset() { + *x = BuildSignal{} + mi := &file_buildsignal_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *BuildSignal) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*BuildSignal) ProtoMessage() {} + +func (x *BuildSignal) ProtoReflect() protoreflect.Message { + mi := &file_buildsignal_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use BuildSignal.ProtoReflect.Descriptor instead. +func (*BuildSignal) Descriptor() ([]byte, []int) { + return file_buildsignal_proto_rawDescGZIP(), []int{0} +} + +func (x *BuildSignal) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +var File_buildsignal_proto protoreflect.FileDescriptor + +const file_buildsignal_proto_rawDesc = "" + + "\n" + + "\x11buildsignal.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\".\n" + + "\vBuildSignal\x12\x0e\n" + + "\x02id\x18\x01 \x01(\tR\x02id:\x0f\x8a\xb5\x18\vbuildsignalB\x82\x01\n" + + "+com.uber.submitqueue.stovepipe.messagequeueB\x10BuildSignalProtoP\x01Z?github.com/uber/submitqueue/stovepipe/core/messagequeue/protopbb\x06proto3" + +var ( + file_buildsignal_proto_rawDescOnce sync.Once + file_buildsignal_proto_rawDescData []byte +) + +func file_buildsignal_proto_rawDescGZIP() []byte { + file_buildsignal_proto_rawDescOnce.Do(func() { + file_buildsignal_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_buildsignal_proto_rawDesc), len(file_buildsignal_proto_rawDesc))) + }) + return file_buildsignal_proto_rawDescData +} + +var file_buildsignal_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_buildsignal_proto_goTypes = []any{ + (*BuildSignal)(nil), // 0: uber.stovepipe.messagequeue.BuildSignal +} +var file_buildsignal_proto_depIdxs = []int32{ + 0, // [0:0] is the sub-list for method output_type + 0, // [0:0] is the sub-list for method input_type + 0, // [0:0] is the sub-list for extension type_name + 0, // [0:0] is the sub-list for extension extendee + 0, // [0:0] is the sub-list for field type_name +} + +func init() { file_buildsignal_proto_init() } +func file_buildsignal_proto_init() { + if File_buildsignal_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_buildsignal_proto_rawDesc), len(file_buildsignal_proto_rawDesc)), + NumEnums: 0, + NumMessages: 1, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_buildsignal_proto_goTypes, + DependencyIndexes: file_buildsignal_proto_depIdxs, + MessageInfos: file_buildsignal_proto_msgTypes, + }.Build() + File_buildsignal_proto = out.File + file_buildsignal_proto_goTypes = nil + file_buildsignal_proto_depIdxs = nil +} diff --git a/stovepipe/core/messagequeue/topics.go b/stovepipe/core/messagequeue/topics.go index 44876958..cbc12387 100644 --- a/stovepipe/core/messagequeue/topics.go +++ b/stovepipe/core/messagequeue/topics.go @@ -28,8 +28,17 @@ const ( // controller consumes it, reloads the Request, and decides the build strategy. TopicKeyProcess TopicKey = "process" - // TopicKeyBuild carries admitted requests from process to build. process - // publishes a BuildRequest (the request id) after it persists the strategy - // and baseline; build reloads the Request from storage. + // TopicKeyBuild carries requests whose build scope has been decided from + // process/analyze to the build stage. The producer publishes a BuildRequest + // (the request id) here; the build controller consumes it, reloads the + // Request, and triggers the build. Partitioned by request id. TopicKeyBuild TopicKey = "build" + + // TopicKeyBuildSignal carries builds to poll from build to the buildsignal + // stage, and from buildsignal back to itself between polls. Producers + // publish a BuildSignal (the build id) here; the buildsignal controller + // consumes it, polls the build runner, and records terminal status. + // Partitioned by build id, so each build's poll loop is an independent + // partition. + TopicKeyBuildSignal TopicKey = "buildsignal" ) diff --git a/tool/proto/BUILD.bazel b/tool/proto/BUILD.bazel index fb86abe2..9e3f55d7 100644 --- a/tool/proto/BUILD.bazel +++ b/tool/proto/BUILD.bazel @@ -68,6 +68,7 @@ go_proto_generated_files( name = "stovepipe_core_messagequeue", srcs = [ "//stovepipe/core/messagequeue/proto:build.proto", + "//stovepipe/core/messagequeue/proto:buildsignal.proto", "//stovepipe/core/messagequeue/proto:process.proto", ], gen_services = False,