diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 8dca48be..3c4b5c4c 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -222,6 +222,7 @@ func run() error { store, queueconfigdefault.NewStore(), fakeSourceControlFactory{}, + registry, stovepipemq.TopicKeyProcess, "stovepipe-process", ) @@ -310,7 +311,7 @@ func run() error { } // newTopicRegistry builds the TopicRegistry for Stovepipe's internal pipeline queues. ingest -// publishes to the process topic and the process consumer subscribes to it. +// publishes to process; process publishes admitted requests to the publish-only build topic. func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) { return consumer.NewTopicRegistry([]consumer.TopicConfig{ { @@ -321,5 +322,10 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe subscriberName, "stovepipe-process", ), }, + { + Key: stovepipemq.TopicKeyBuild, + Name: "build", + Queue: q, + }, }) } diff --git a/stovepipe/controller/process/BUILD.bazel b/stovepipe/controller/process/BUILD.bazel index b55513c1..2ff96fdd 100644 --- a/stovepipe/controller/process/BUILD.bazel +++ b/stovepipe/controller/process/BUILD.bazel @@ -6,6 +6,7 @@ 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", diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index 1dce57c1..a99f6503 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -24,6 +24,7 @@ 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" @@ -44,6 +45,7 @@ type Controller struct { store storage.Storage queueConfigs queueconfig.Store sourceControl sourcecontrol.Factory + registry consumer.TopicRegistry topicKey consumer.TopicKey consumerGroup string } @@ -61,6 +63,7 @@ func NewController( store storage.Storage, queueConfigs queueconfig.Store, sourceControl sourcecontrol.Factory, + registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) *Controller { @@ -70,6 +73,7 @@ func NewController( store: store, queueConfigs: queueConfigs, sourceControl: sourceControl, + registry: registry, topicKey: topicKey, consumerGroup: consumerGroup, } @@ -98,7 +102,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r switch request.State { case entity.RequestStateProcessing: - // TODO: re-publish to build once the build stage lands (RFC process algorithm, step 3). + if err := c.publishBuild(ctx, request.ID); err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1) + return fmt.Errorf("ProcessController failed to publish request %s to build: %w", request.ID, err) + } return nil case entity.RequestStateSuperseded: return nil @@ -238,7 +245,10 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request return nil } - // TODO(build-publish): publish BuildRequest to the build stage here. + if err := c.publishBuild(ctx, request.ID); err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1) + return fmt.Errorf("ProcessController failed to publish request %s to build: %w", request.ID, err) + } metrics.NamedCounter(c.metricsScope, _opName, "admitted", 1, metrics.NewTag("strategy", string(request.BuildStrategy)), @@ -432,6 +442,29 @@ func (c *Controller) loadQueue(ctx context.Context, name string) (entity.Queue, return entity.Queue{}, fmt.Errorf("ProcessController failed to load queue %s: %w", name, err) } +// 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. +func (c *Controller) publishBuild(ctx context.Context, id string) error { + payload, err := stovepipemq.Marshal(&stovepipemq.BuildRequest{Id: id}) + 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 { + return fmt.Errorf("failed to publish build request: %w", err) + } + return nil +} + // Name returns the controller name for logging and metrics. func (c *Controller) Name() string { return "process" diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index 13008712..05461576 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -49,6 +49,7 @@ type processMocks struct { queueStore *storagemock.MockQueueStore sourceFactory *sourcecontrolmock.MockFactory sourceControl *sourcecontrolmock.MockSourceControl + publisher *mqmock.MockPublisher } func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processMocks) { @@ -63,18 +64,27 @@ func newControllerWithScope(t *testing.T, ctrl *gomock.Controller, scope tally.S queueStore: storagemock.NewMockQueueStore(ctrl), sourceFactory: sourcecontrolmock.NewMockFactory(ctrl), sourceControl: sourcecontrolmock.NewMockSourceControl(ctrl), + publisher: mqmock.NewMockPublisher(ctrl), } store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() + queue := mqmock.NewMockQueue(ctrl) + queue.EXPECT().Publisher().Return(m.publisher).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: stovepipemq.TopicKeyBuild, Name: "build", Queue: queue}, + }) + require.NoError(t, err) + c := NewController( zap.NewNop().Sugar(), scope, store, queueconfigdefault.NewStore(), m.sourceFactory, + registry, stovepipemq.TopicKeyProcess, "stovepipe-process", ) @@ -106,7 +116,24 @@ func acceptedRequest(id string) entity.Request { } } -func expectAdmit(m processMocks, id string) { +func TestProcessBuildPublishRequiresRegisteredTopic(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + c.registry = consumer.TopicRegistry{} + + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{ + ID: testID, Queue: testQueue, State: entity.RequestStateProcessing, Version: 2, + }, nil) + + err := c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID))) + + require.Error(t, err) + assert.False(t, errs.IsRetryable(err)) +} + +func expectAdmit(t *testing.T, m processMocks, id string) { + t.Helper() + updatedQueue := entity.Queue{ Name: testQueue, LatestRequestID: id, @@ -119,6 +146,23 @@ func expectAdmit(m processMocks, id string) { updatedReq.State = entity.RequestStateProcessing updatedReq.BuildStrategy = entity.BuildStrategyFull m.reqStore.EXPECT().Update(gomock.Any(), updatedReq, int32(1), int32(2)).Return(nil) + + expectBuildPublish(t, m, id) +} + +func expectBuildPublish(t *testing.T, m processMocks, id string) { + t.Helper() + + m.publisher.EXPECT(). + Publish(gomock.Any(), "build", gomock.AssignableToTypeOf(entityqueue.Message{})). + DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { + assert.Equal(t, id, msg.ID) + assert.Equal(t, id, msg.PartitionKey) + buildReq := &stovepipemq.BuildRequest{} + require.NoError(t, stovepipemq.Unmarshal(msg.Payload, buildReq)) + assert.Equal(t, id, buildReq.Id) + return nil + }) } func TestDeriveBuildStrategy(t *testing.T) { @@ -257,7 +301,7 @@ func TestProcessEmitsAdmittedStrategyMetric(t *testing.T) { LatestRequestID: testID, Version: 1, }, nil) - expectAdmit(m, testID) + expectAdmit(t, m, testID) require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID)))) @@ -349,6 +393,7 @@ func TestProcessRederivesStrategyAfterQueueReload(t *testing.T) { m.sourceControl.EXPECT().IsAncestor(gomock.Any(), reloadedLastGreen, testURI).Return(true, nil) m.queueStore.EXPECT().Update(gomock.Any(), claimedQueue, int32(2), int32(3)).Return(nil) m.reqStore.EXPECT().Update(gomock.Any(), updatedRequest, int32(1), int32(2)).Return(nil) + expectBuildPublish(t, m, testID) require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID)))) }) @@ -373,11 +418,12 @@ func TestProcess(t *testing.T) { }, }, { - name: "processing is no-op until build publish lands", + name: "processing republishes to build", setup: func(m processMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{ ID: testID, Queue: testQueue, State: entity.RequestStateProcessing, Version: 2, }, nil) + expectBuildPublish(t, m, testID) }, }, { @@ -397,7 +443,33 @@ func TestProcess(t *testing.T) { LatestRequestID: testID, Version: 1, }, nil) - expectAdmit(m, testID) + expectAdmit(t, m, testID) + }, + }, + { + name: "build publish failure retains admitted request and claimed slot", + wantErr: true, + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + Version: 1, + }, nil) + updatedQueue := entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + InFlightCount: 1, + Version: 1, + } + m.queueStore.EXPECT().Update(gomock.Any(), updatedQueue, int32(1), int32(2)).Return(nil) + updatedRequest := acceptedRequest(testID) + updatedRequest.State = entity.RequestStateProcessing + updatedRequest.BuildStrategy = entity.BuildStrategyFull + m.reqStore.EXPECT().Update(gomock.Any(), updatedRequest, int32(1), int32(2)).Return(nil) + m.publisher.EXPECT(). + Publish(gomock.Any(), "build", gomock.AssignableToTypeOf(entityqueue.Message{})). + Return(errors.New("queue unavailable")) }, }, { @@ -462,6 +534,7 @@ func TestProcess(t *testing.T) { updatedReq.State = entity.RequestStateProcessing updatedReq.BuildStrategy = entity.BuildStrategyFull m.reqStore.EXPECT().Update(gomock.Any(), updatedReq, int32(1), int32(2)).Return(nil) + expectBuildPublish(t, m, testID) }, }, { @@ -529,6 +602,7 @@ func TestProcess(t *testing.T) { retry.BuildStrategy = entity.BuildStrategyIncrementalSinceGreen retry.BaseURI = lastGreenURI m.reqStore.EXPECT().Update(gomock.Any(), retry, int32(2), int32(3)).Return(nil) + expectBuildPublish(t, m, testID) }, }, { diff --git a/stovepipe/core/messagequeue/messagequeue.go b/stovepipe/core/messagequeue/messagequeue.go index f8e03fb6..e88933e1 100644 --- a/stovepipe/core/messagequeue/messagequeue.go +++ b/stovepipe/core/messagequeue/messagequeue.go @@ -41,6 +41,10 @@ type ( // ProcessRequest is the payload ingest publishes to the process stage: the // 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 = protopb.BuildRequest ) // 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/process_test.go index 4c9345f3..185de8de 100644 --- a/stovepipe/core/messagequeue/process_test.go +++ b/stovepipe/core/messagequeue/process_test.go @@ -33,6 +33,17 @@ func TestProcessRequestRoundTrip(t *testing.T) { assert.True(t, proto.Equal(req, got), "round-tripped ProcessRequest should equal the original") } +func TestBuildRequestRoundTrip(t *testing.T) { + req := &BuildRequest{Id: "request/monorepo/main/42"} + + data, err := Marshal(req) + require.NoError(t, err) + + got := &BuildRequest{} + require.NoError(t, Unmarshal(data, got)) + assert.True(t, proto.Equal(req, got), "round-tripped BuildRequest should equal the original") +} + // TestWireFormat locks the protojson encoding decision the contract relies on: // snake_case field names (UseProtoNames). func TestWireFormat(t *testing.T) { @@ -47,7 +58,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{}} { + for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}} { keys := TopicKeys(m) require.NotEmpty(t, keys, "message must declare a non-empty topic_keys option") for _, key := range keys { @@ -57,6 +68,7 @@ func TestTopicKeysBindEveryTopicKey(t *testing.T) { keys := []TopicKey{ TopicKeyProcess, + TopicKeyBuild, } valid := map[string]bool{} diff --git a/stovepipe/core/messagequeue/proto/BUILD.bazel b/stovepipe/core/messagequeue/proto/BUILD.bazel index 3a783684..95634720 100644 --- a/stovepipe/core/messagequeue/proto/BUILD.bazel +++ b/stovepipe/core/messagequeue/proto/BUILD.bazel @@ -1,4 +1,7 @@ exports_files( - ["process.proto"], + [ + "build.proto", + "process.proto", + ], visibility = ["//tool/proto:__pkg__"], ) diff --git a/stovepipe/core/messagequeue/proto/build.proto b/stovepipe/core/messagequeue/proto/build.proto new file mode 100644 index 00000000..5b607614 --- /dev/null +++ b/stovepipe/core/messagequeue/proto/build.proto @@ -0,0 +1,34 @@ +// 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 = "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. +message BuildRequest { + option (uber.base.messagequeue.topic_keys) = "build"; + + // id is the admitted request id to build. Format: "request//". + string id = 1; +} diff --git a/stovepipe/core/messagequeue/protopb/BUILD.bazel b/stovepipe/core/messagequeue/protopb/BUILD.bazel index 0f7c020e..9a16d825 100644 --- a/stovepipe/core/messagequeue/protopb/BUILD.bazel +++ b/stovepipe/core/messagequeue/protopb/BUILD.bazel @@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_library") go_library( name = "go_default_library", - srcs = ["process.pb.go"], + srcs = [ + "build.pb.go", + "process.pb.go", + ], importpath = "github.com/uber/submitqueue/stovepipe/core/messagequeue/protopb", visibility = ["//visibility:public"], deps = [ diff --git a/stovepipe/core/messagequeue/protopb/build.pb.go b/stovepipe/core/messagequeue/protopb/build.pb.go new file mode 100644 index 00000000..60025cf8 --- /dev/null +++ b/stovepipe/core/messagequeue/protopb/build.pb.go @@ -0,0 +1,144 @@ +// 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: build.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) +) + +// 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. +type BuildRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // id is the admitted request id to build. Format: "request//". + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *BuildRequest) Reset() { + *x = BuildRequest{} + mi := &file_build_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *BuildRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*BuildRequest) ProtoMessage() {} + +func (x *BuildRequest) ProtoReflect() protoreflect.Message { + mi := &file_build_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 BuildRequest.ProtoReflect.Descriptor instead. +func (*BuildRequest) Descriptor() ([]byte, []int) { + return file_build_proto_rawDescGZIP(), []int{0} +} + +func (x *BuildRequest) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +var File_build_proto protoreflect.FileDescriptor + +const file_build_proto_rawDesc = "" + + "\n" + + "\vbuild.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\")\n" + + "\fBuildRequest\x12\x0e\n" + + "\x02id\x18\x01 \x01(\tR\x02id:\t\x8a\xb5\x18\x05buildB|\n" + + "+com.uber.submitqueue.stovepipe.messagequeueB\n" + + "BuildProtoP\x01Z?github.com/uber/submitqueue/stovepipe/core/messagequeue/protopbb\x06proto3" + +var ( + file_build_proto_rawDescOnce sync.Once + file_build_proto_rawDescData []byte +) + +func file_build_proto_rawDescGZIP() []byte { + file_build_proto_rawDescOnce.Do(func() { + file_build_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_build_proto_rawDesc), len(file_build_proto_rawDesc))) + }) + return file_build_proto_rawDescData +} + +var file_build_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_build_proto_goTypes = []any{ + (*BuildRequest)(nil), // 0: uber.stovepipe.messagequeue.BuildRequest +} +var file_build_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_build_proto_init() } +func file_build_proto_init() { + if File_build_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_build_proto_rawDesc), len(file_build_proto_rawDesc)), + NumEnums: 0, + NumMessages: 1, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_build_proto_goTypes, + DependencyIndexes: file_build_proto_depIdxs, + MessageInfos: file_build_proto_msgTypes, + }.Build() + File_build_proto = out.File + file_build_proto_goTypes = nil + file_build_proto_depIdxs = nil +} diff --git a/stovepipe/core/messagequeue/topics.go b/stovepipe/core/messagequeue/topics.go index f4a0109c..44876958 100644 --- a/stovepipe/core/messagequeue/topics.go +++ b/stovepipe/core/messagequeue/topics.go @@ -27,4 +27,9 @@ const ( // stage. ingest publishes a ProcessRequest (the request id) here; the process // 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 TopicKey = "build" ) diff --git a/tool/proto/BUILD.bazel b/tool/proto/BUILD.bazel index f710f7c0..fb86abe2 100644 --- a/tool/proto/BUILD.bazel +++ b/tool/proto/BUILD.bazel @@ -66,7 +66,10 @@ go_proto_generated_files( # Stovepipe internal queue contract (message-only, no RPC service). go_proto_generated_files( name = "stovepipe_core_messagequeue", - srcs = ["//stovepipe/core/messagequeue/proto:process.proto"], + srcs = [ + "//stovepipe/core/messagequeue/proto:build.proto", + "//stovepipe/core/messagequeue/proto:process.proto", + ], gen_services = False, imports = [ "//api/base/messagequeue/proto:messagequeue.proto",