diff --git a/stovepipe/controller/fix/BUILD.bazel b/stovepipe/controller/fix/BUILD.bazel index a89b8a27a..0d8d045d1 100644 --- a/stovepipe/controller/fix/BUILD.bazel +++ b/stovepipe/controller/fix/BUILD.bazel @@ -2,12 +2,17 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", - srcs = ["fix.go"], + srcs = [ + "delivery.go", + "fix.go", + ], importpath = "github.com/uber/submitqueue/stovepipe/controller/fix", visibility = ["//visibility:public"], deps = [ + "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/fixer:go_default_library", "//stovepipe/extension/storage:go_default_library", @@ -18,11 +23,17 @@ go_library( go_test( name = "go_default_test", - srcs = ["fix_test.go"], + srcs = [ + "delivery_test.go", + "fix_test.go", + ], embed = [":go_default_library"], deps = [ + "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/fixer:go_default_library", "//stovepipe/extension/fixer/mock:go_default_library", diff --git a/stovepipe/controller/fix/README.md b/stovepipe/controller/fix/README.md index 4e2cbfba1..bca017ade 100644 --- a/stovepipe/controller/fix/README.md +++ b/stovepipe/controller/fix/README.md @@ -1,7 +1,9 @@ # Fix submission -The controller pages through a build's durable issue records and skips submission when any record has a fix ID, including recovered issues. Otherwise, it selects one currently open issue whose first-seen build matches this build and calls `RepairExecutor.Submit`. +The controller consumes a `FixRequest` containing a queue name and build ID. It pages through a build's durable issue records and skips submission when any record has a fix ID, including recovered issues. Otherwise, it selects one currently open issue whose first-seen build matches this build and calls `RepairExecutor.Submit`. A returned fix ID is associated with the issue through `BuildIssueStore.Update`, using the version read during selection. A concurrent association is preserved. A submission error without an ID leaves the issue eligible for normal queue retries; an ID returned alongside an error is retained. +Issue tracking calls [`fixrequest.Publish`](../../core/fixrequest/publish.go) after persisting both the build observation and open issue. Publications for the same build share a message identity and partition. + The one-per-build limit is controller policy. Submission precedes persistence, so overlapping deliveries, crashes, and failed writes can still produce duplicate executor calls. Issue recovery remains separate from repair association. diff --git a/stovepipe/controller/fix/delivery.go b/stovepipe/controller/fix/delivery.go new file mode 100644 index 000000000..d07ea2b5d --- /dev/null +++ b/stovepipe/controller/fix/delivery.go @@ -0,0 +1,41 @@ +// Copyright (c) 2026 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 fix + +import ( + "context" + "fmt" + + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + mq "github.com/uber/submitqueue/stovepipe/core/messagequeue" +) + +var _ consumer.Controller = (*Controller)(nil) + +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { + message := delivery.Message() + request := &mq.FixRequest{} + if err := mq.Unmarshal(message.Payload, request); err != nil { + return fmt.Errorf("failed to deserialize repair request: %w", err) + } + if err := entityqueue.ValidatePayloadQueue(message, request.GetQueueName()); err != nil { + return fmt.Errorf("invalid message identity: %w", err) + } + if request.GetId() == "" { + return fmt.Errorf("repair request requires a build ID") + } + return c.processBuild(ctx, request.GetQueueName(), request.GetId()) +} diff --git a/stovepipe/controller/fix/delivery_test.go b/stovepipe/controller/fix/delivery_test.go new file mode 100644 index 000000000..43efff41c --- /dev/null +++ b/stovepipe/controller/fix/delivery_test.go @@ -0,0 +1,80 @@ +// Copyright (c) 2026 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 fix + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" + "github.com/uber/submitqueue/platform/errs" + mq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "go.uber.org/mock/gomock" +) + +func repairMessage(t *testing.T) entityqueue.Message { + t.Helper() + payload, err := mq.Marshal(&mq.FixRequest{Id: testBuildID, QueueName: testQueue}) + require.NoError(t, err) + message := entityqueue.NewMessage(testBuildID, payload, testBuildID, nil) + message.Tenant = testQueue + return message +} + +func (m controllerMocks) delivery(message entityqueue.Message) *consumermock.MockDelivery { + delivery := consumermock.NewMockDelivery(m.ctrl) + delivery.EXPECT().Message().Return(message) + return delivery +} + +func TestProcessSubmitsBuildIssue(t *testing.T) { + controller, m := newTestController(t) + issue := testIssue() + m.expectIssue(issue) + m.executor.EXPECT().Submit(gomock.Any(), issue).Return("attempt", nil) + m.observations.EXPECT().Update(gomock.Any(), testBuildID, issueWithFixID(issue, "attempt"), int32(0), int32(1)).Return(nil) + + require.NoError(t, controller.Process(context.Background(), m.delivery(repairMessage(t)))) +} + +func TestProcessRejectsInvalidMessagesBeforeListingIssues(t *testing.T) { + for _, kind := range []string{"invalid JSON", "empty build", "empty queue", "empty tenant", "wrong tenant", "wrong metadata"} { + t.Run(kind, func(t *testing.T) { + controller, m := newTestController(t) + message := repairMessage(t) + switch kind { + case "invalid JSON": + message.Payload = []byte("{") + case "empty build": + message.Payload = []byte(`{"queue_name":"validation/main"}`) + case "empty queue": + message.Payload = []byte(`{"id":"Build/42 "}`) + case "empty tenant": + message.Tenant = "" + case "wrong tenant": + message.Tenant = "other/main" + case "wrong metadata": + message.Metadata[entityqueue.MetadataKeyQueueName] = "other/main" + } + + err := controller.Process(context.Background(), m.delivery(message)) + require.Error(t, err) + assert.False(t, errs.IsRetryable(err)) + }) + } +} diff --git a/stovepipe/core/fixrequest/BUILD.bazel b/stovepipe/core/fixrequest/BUILD.bazel new file mode 100644 index 000000000..75ae5b8ae --- /dev/null +++ b/stovepipe/core/fixrequest/BUILD.bazel @@ -0,0 +1,28 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["publish.go"], + importpath = "github.com/uber/submitqueue/stovepipe/core/fixrequest", + visibility = ["//visibility:public"], + deps = [ + "//platform/consumer:go_default_library", + "//platform/publish:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["publish_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/extension/messagequeue/mock:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + ], +) diff --git a/stovepipe/core/fixrequest/publish.go b/stovepipe/core/fixrequest/publish.go new file mode 100644 index 000000000..21f684316 --- /dev/null +++ b/stovepipe/core/fixrequest/publish.go @@ -0,0 +1,46 @@ +// Copyright (c) 2026 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 fixrequest publishes builds for repair submission. +package fixrequest + +import ( + "context" + "fmt" + + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/publish" + mq "github.com/uber/submitqueue/stovepipe/core/messagequeue" +) + +// Publish requires the build and its open issues to be persisted first. +// Replays after publication errors must call Publish again. +func Publish(ctx context.Context, registry consumer.TopicRegistry, queueName, buildID string) error { + if queueName == "" || buildID == "" { + return fmt.Errorf("repair request requires a queue and build ID") + } + + payload, err := mq.Marshal(&mq.FixRequest{Id: buildID, QueueName: queueName}) + if err != nil { + return fmt.Errorf("failed to serialize repair request: %w", err) + } + + // Repair submission is a one-shot handoff for this build. + return publish.Message(ctx, registry, mq.TopicKeyFix, publish.MessageParams{ + Tenant: queueName, + ID: publish.IntentID(buildID), + Payload: payload, + PartitionKey: buildID, + }) +} diff --git a/stovepipe/core/fixrequest/publish_test.go b/stovepipe/core/fixrequest/publish_test.go new file mode 100644 index 000000000..0b2a600cc --- /dev/null +++ b/stovepipe/core/fixrequest/publish_test.go @@ -0,0 +1,95 @@ +// Copyright (c) 2026 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 fixrequest + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + mq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "go.uber.org/mock/gomock" +) + +func newTestRegistry(t *testing.T) (consumer.TopicRegistry, *queuemock.MockPublisher) { + t.Helper() + ctrl := gomock.NewController(t) + queue := queuemock.NewMockQueue(ctrl) + publisher := queuemock.NewMockPublisher(ctrl) + queue.EXPECT().Publisher().Return(publisher).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: mq.TopicKeyFix, Name: "test-fix", Queue: queue}, + }) + require.NoError(t, err) + return registry, publisher +} + +func TestPublishBuildIdentityAndRetry(t *testing.T) { + registry, publisher := newTestRegistry(t) + var messages []entityqueue.Message + publisher.EXPECT().Publish(gomock.Any(), "test-fix", gomock.Any()). + DoAndReturn(func(_ context.Context, _ string, message entityqueue.Message) error { + messages = append(messages, message) + if len(messages) == 1 { + return assert.AnError + } + return nil + }).Times(4) + + ctx := context.Background() + require.ErrorIs(t, Publish(ctx, registry, "validation/main", "Build/42 "), assert.AnError) + require.NoError(t, Publish(ctx, registry, "validation/main", "Build/42 ")) + require.NoError(t, Publish(ctx, registry, "validation/main", "Build/43")) + require.NoError(t, Publish(ctx, registry, "validation/other", "Build/42 ")) + + assert.Equal(t, messages[0].ID, messages[1].ID) + assert.Equal(t, messages[0].PartitionKey, messages[1].PartitionKey) + assert.NotEqual(t, messages[1].ID, messages[2].ID) + assert.NotEqual(t, messages[1].Tenant, messages[3].Tenant) + for _, message := range messages { + payload := &mq.FixRequest{} + require.NoError(t, mq.Unmarshal(message.Payload, payload)) + assert.Equal(t, payload.GetId(), message.PartitionKey) + assert.Equal(t, payload.GetQueueName(), message.Tenant) + assert.Equal(t, message.Tenant, message.Metadata[entityqueue.MetadataKeyQueueName]) + } + assert.Equal(t, "Build/42 ", messages[0].PartitionKey) +} + +func TestPublishRejectsMissingIdentity(t *testing.T) { + for _, test := range []struct { + name string + queue string + build string + }{ + {name: "missing queue", build: "build/42"}, + {name: "missing build", queue: "validation/main"}, + } { + t.Run(test.name, func(t *testing.T) { + registry, _ := newTestRegistry(t) + require.Error(t, Publish(context.Background(), registry, test.queue, test.build)) + }) + } +} + +func TestPublishReturnsConfigurationAndSerializationErrors(t *testing.T) { + require.Error(t, Publish(context.Background(), consumer.TopicRegistry{}, "validation/main", "build/42")) + registry, _ := newTestRegistry(t) + require.Error(t, Publish(context.Background(), registry, "validation/main", string([]byte{0xff}))) +} diff --git a/stovepipe/core/messagequeue/messagequeue.go b/stovepipe/core/messagequeue/messagequeue.go index e885cb255..487d88cb6 100644 --- a/stovepipe/core/messagequeue/messagequeue.go +++ b/stovepipe/core/messagequeue/messagequeue.go @@ -56,6 +56,9 @@ type ( // IssueTrack identifies persisted build observations for queue-scoped reconciliation. IssueTrack = protopb.IssueTrack + // FixRequest identifies the build whose issues should be considered for repair. + FixRequest = protopb.FixRequest + // Record is the payload buildsignal publishes to the record stage once a // build reaches a terminal status: the build id to record. Record = protopb.Record diff --git a/stovepipe/core/messagequeue/messagequeue_test.go b/stovepipe/core/messagequeue/messagequeue_test.go index cd58c565e..fd3e20ac2 100644 --- a/stovepipe/core/messagequeue/messagequeue_test.go +++ b/stovepipe/core/messagequeue/messagequeue_test.go @@ -80,7 +80,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{}, &BuildSignal{}, &BuildAnalysis{}, &IssueTrack{}, &Record{}} { + for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}, &BuildAnalysis{}, &IssueTrack{}, &FixRequest{}, &Record{}} { keys := TopicKeys(m) require.NotEmpty(t, keys, "message must declare a non-empty topic_keys option") for _, key := range keys { @@ -94,6 +94,7 @@ func TestTopicKeysBindEveryTopicKey(t *testing.T) { TopicKeyBuildSignal, TopicKeyBuildAnalysis, TopicKeyIssueTrack, + TopicKeyFix, TopicKeyRecord, } @@ -146,3 +147,15 @@ func TestIssueTrackToleratesUnknownFields(t *testing.T) { require.NoError(t, Unmarshal([]byte(`{"build_id":"build/42","queue_name":"validation/main","finalize":true,"future_field":42}`), got)) assert.True(t, proto.Equal(&IssueTrack{BuildId: "build/42", QueueName: "validation/main", Finalize: true}, got)) } + +func TestFixRequestRoundTrip(t *testing.T) { + request := &FixRequest{Id: "Build/42 ", QueueName: "validation/main"} + data, err := Marshal(request) + require.NoError(t, err) + + got := &FixRequest{} + require.NoError(t, Unmarshal(data, got)) + + assert.True(t, proto.Equal(request, got)) + assert.JSONEq(t, `{"id":"Build/42 ","queue_name":"validation/main"}`, string(data)) +} diff --git a/stovepipe/core/messagequeue/proto/BUILD.bazel b/stovepipe/core/messagequeue/proto/BUILD.bazel index e942b7f1f..162570f78 100644 --- a/stovepipe/core/messagequeue/proto/BUILD.bazel +++ b/stovepipe/core/messagequeue/proto/BUILD.bazel @@ -3,6 +3,7 @@ exports_files( "build.proto", "buildanalysis.proto", "buildsignal.proto", + "fix.proto", "issuetrack.proto", "process.proto", "record.proto", diff --git a/stovepipe/core/messagequeue/proto/fix.proto b/stovepipe/core/messagequeue/proto/fix.proto new file mode 100644 index 000000000..a9e9a4a4e --- /dev/null +++ b/stovepipe/core/messagequeue/proto/fix.proto @@ -0,0 +1,34 @@ +// Copyright (c) 2026 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 = "FixProto"; +option java_package = "com.uber.submitqueue.stovepipe.messagequeue"; + +// FixRequest identifies a persisted build whose issues should be considered for repair. +message FixRequest { + option (uber.base.messagequeue.topic_keys) = "fix"; + + // queue_name is the required queue namespace of the build. + string queue_name = 2; + // id is the opaque runner-assigned build identifier. + string id = 1; +} diff --git a/stovepipe/core/messagequeue/protopb/BUILD.bazel b/stovepipe/core/messagequeue/protopb/BUILD.bazel index 61c31ef92..177796fe9 100644 --- a/stovepipe/core/messagequeue/protopb/BUILD.bazel +++ b/stovepipe/core/messagequeue/protopb/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "build.pb.go", "buildanalysis.pb.go", "buildsignal.pb.go", + "fix.pb.go", "issuetrack.pb.go", "process.pb.go", "record.pb.go", diff --git a/stovepipe/core/messagequeue/protopb/fix.pb.go b/stovepipe/core/messagequeue/protopb/fix.pb.go new file mode 100644 index 000000000..30f38b5ae --- /dev/null +++ b/stovepipe/core/messagequeue/protopb/fix.pb.go @@ -0,0 +1,153 @@ +// Copyright (c) 2026 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.11 +// protoc v5.29.3 +// source: fix.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) +) + +// FixRequest identifies a persisted build whose issues should be considered for repair. +type FixRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // queue_name is the required queue namespace of the build. + QueueName string `protobuf:"bytes,2,opt,name=queue_name,json=queueName,proto3" json:"queue_name,omitempty"` + // id is the opaque runner-assigned build identifier. + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *FixRequest) Reset() { + *x = FixRequest{} + mi := &file_fix_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *FixRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*FixRequest) ProtoMessage() {} + +func (x *FixRequest) ProtoReflect() protoreflect.Message { + mi := &file_fix_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 FixRequest.ProtoReflect.Descriptor instead. +func (*FixRequest) Descriptor() ([]byte, []int) { + return file_fix_proto_rawDescGZIP(), []int{0} +} + +func (x *FixRequest) GetQueueName() string { + if x != nil { + return x.QueueName + } + return "" +} + +func (x *FixRequest) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +var File_fix_proto protoreflect.FileDescriptor + +const file_fix_proto_rawDesc = "" + + "\n" + + "\tfix.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"D\n" + + "\n" + + "FixRequest\x12\x1d\n" + + "\n" + + "queue_name\x18\x02 \x01(\tR\tqueueName\x12\x0e\n" + + "\x02id\x18\x01 \x01(\tR\x02id:\a\x8a\xb5\x18\x03fixBz\n" + + "+com.uber.submitqueue.stovepipe.messagequeueB\bFixProtoP\x01Z?github.com/uber/submitqueue/stovepipe/core/messagequeue/protopbb\x06proto3" + +var ( + file_fix_proto_rawDescOnce sync.Once + file_fix_proto_rawDescData []byte +) + +func file_fix_proto_rawDescGZIP() []byte { + file_fix_proto_rawDescOnce.Do(func() { + file_fix_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_fix_proto_rawDesc), len(file_fix_proto_rawDesc))) + }) + return file_fix_proto_rawDescData +} + +var file_fix_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_fix_proto_goTypes = []any{ + (*FixRequest)(nil), // 0: uber.stovepipe.messagequeue.FixRequest +} +var file_fix_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_fix_proto_init() } +func file_fix_proto_init() { + if File_fix_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_fix_proto_rawDesc), len(file_fix_proto_rawDesc)), + NumEnums: 0, + NumMessages: 1, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_fix_proto_goTypes, + DependencyIndexes: file_fix_proto_depIdxs, + MessageInfos: file_fix_proto_msgTypes, + }.Build() + File_fix_proto = out.File + file_fix_proto_goTypes = nil + file_fix_proto_depIdxs = nil +} diff --git a/stovepipe/core/messagequeue/topics.go b/stovepipe/core/messagequeue/topics.go index 433bd4fe9..5571f917d 100644 --- a/stovepipe/core/messagequeue/topics.go +++ b/stovepipe/core/messagequeue/topics.go @@ -48,6 +48,9 @@ const ( // TopicKeyIssueTrack carries persisted observation notifications, partitioned by queue. TopicKeyIssueTrack TopicKey = "issuetrack" + // TopicKeyFix carries build IDs for repair submission, partitioned by build ID. + TopicKeyFix TopicKey = "fix" + // TopicKeyRecord carries a request whose build reached a terminal status from // buildsignal to the record stage. The buildsignal controller publishes a // Record (the request id) here once, and only once, a build reaches a diff --git a/tool/proto/BUILD.bazel b/tool/proto/BUILD.bazel index 8747096b1..8c9b8a40b 100644 --- a/tool/proto/BUILD.bazel +++ b/tool/proto/BUILD.bazel @@ -105,6 +105,7 @@ go_proto_generated_files( "//stovepipe/core/messagequeue/proto:build.proto", "//stovepipe/core/messagequeue/proto:buildanalysis.proto", "//stovepipe/core/messagequeue/proto:buildsignal.proto", + "//stovepipe/core/messagequeue/proto:fix.proto", "//stovepipe/core/messagequeue/proto:issuetrack.proto", "//stovepipe/core/messagequeue/proto:process.proto", "//stovepipe/core/messagequeue/proto:record.proto",