From 1577cd21e03a030ab3452ad351273263e0c5ed3c Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Fri, 23 Oct 2026 04:19:08 +0000 Subject: [PATCH] feat(stovepipe): Define issue-tracking handoff --- stovepipe/core/messagequeue/README.md | 10 ++ stovepipe/core/messagequeue/messagequeue.go | 3 + .../core/messagequeue/messagequeue_test.go | 31 +++- stovepipe/core/messagequeue/proto/BUILD.bazel | 1 + .../core/messagequeue/proto/issuetrack.proto | 39 ++++ .../core/messagequeue/protopb/BUILD.bazel | 1 + .../messagequeue/protopb/issuetrack.pb.go | 167 ++++++++++++++++++ stovepipe/core/messagequeue/topics.go | 3 + stovepipe/entity/build_issue.go | 2 +- stovepipe/extension/storage/README.md | 4 +- .../extension/storage/build_issue_store.go | 22 +++ .../storage/mock/build_issue_store_mock.go | 16 ++ stovepipe/extension/storage/mysql/BUILD.bazel | 1 + .../storage/mysql/build_issue_list_test.go | 138 +++++++++++++++ .../storage/mysql/build_issue_store.go | 67 ++++++- .../storage/mysql/build_issue_store_test.go | 25 +++ .../extension/storage/mysql/BUILD.bazel | 1 + .../storage/mysql/build_issue_list_test.go | 156 ++++++++++++++++ .../extension/storage/mysql/storage_test.go | 5 + tool/proto/BUILD.bazel | 1 + 20 files changed, 688 insertions(+), 5 deletions(-) create mode 100644 stovepipe/core/messagequeue/proto/issuetrack.proto create mode 100644 stovepipe/core/messagequeue/protopb/issuetrack.pb.go create mode 100644 stovepipe/extension/storage/mysql/build_issue_list_test.go create mode 100644 test/integration/stovepipe/extension/storage/mysql/build_issue_list_test.go diff --git a/stovepipe/core/messagequeue/README.md b/stovepipe/core/messagequeue/README.md index 39d8df7a8..a4f05b33d 100644 --- a/stovepipe/core/messagequeue/README.md +++ b/stovepipe/core/messagequeue/README.md @@ -11,3 +11,13 @@ Payloads are defined in proto3 (`proto/`, generated into `protopb/`) and seriali - **buildsignal** (`TopicKeyBuildSignal`, `BuildSignal`) — build publishes the build id here after triggering; buildsignal holds the delivery between polls until the build reaches a terminal status. Partitioned by build id, so each build's poll loop is an independent partition. - **buildanalysis** (`TopicKeyBuildAnalysis`, `BuildAnalysis`) — identifies a persisted build for issue detection, partitioned by build id. Build publishes a start notice and buildsignal publishes a completion notice with a distinct message id. See [build analysis](../../controller/buildanalysis/README.md) for polling and persistence behavior. - **record** (`TopicKeyRecord`, `Record`) — buildsignal publishes the request id after persisting the terminal outcome; record reloads the request and records its validation fact. Partitioned by request id. + +## Issue tracking handoff + +`IssueTrack` binds to `TopicKeyIssueTrack` and identifies a persisted build and its queue. Notifications are partitioned by queue so both running and final issue-state updates share one serialized path, independently of build-partitioned analysis polling. The message carries identifiers rather than a failure collection; per-build observations are read through `BuildIssueStore.List`. + +Publish only after detection and observation persistence succeed. With `finalize` unset, the notification permits opening newly detected issues, not resolution by absence. With `finalize` set, it requests final reconciliation after a successful terminal analysis. Publish that notification even when no failures were detected; terminal Build status alone does not replace the handoff. Publication retries must also follow duplicate observation writes so a crash between persistence and publication does not strand work. + +Per-build observations accumulate historical evidence; finalization does not turn them into a replacement snapshot or override the detector's absence semantics. The reconciler must establish the authoritative final failure set and applicable validation scope before resolving issues, reject older requests, and ignore running notifications after finalization. Notification identity must distinguish new observations from finalization while remaining stable on publication retries. Business behavior follows the payload, not a parsed message ID. + +Only the handoff contract is defined. No publisher, consumer, topic registration, reconciliation progress, or writes to open issues and lifecycle history are enabled. diff --git a/stovepipe/core/messagequeue/messagequeue.go b/stovepipe/core/messagequeue/messagequeue.go index e9fb2c944..e885cb255 100644 --- a/stovepipe/core/messagequeue/messagequeue.go +++ b/stovepipe/core/messagequeue/messagequeue.go @@ -53,6 +53,9 @@ type ( // BuildAnalysis identifies the build whose issue observations are polled. BuildAnalysis = protopb.BuildAnalysis + // IssueTrack identifies persisted build observations for queue-scoped reconciliation. + IssueTrack = protopb.IssueTrack + // 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 cbcc1c884..cd58c565e 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{}, &Record{}} { + for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}, &BuildAnalysis{}, &IssueTrack{}, &Record{}} { keys := TopicKeys(m) require.NotEmpty(t, keys, "message must declare a non-empty topic_keys option") for _, key := range keys { @@ -93,6 +93,7 @@ func TestTopicKeysBindEveryTopicKey(t *testing.T) { TopicKeyBuild, TopicKeyBuildSignal, TopicKeyBuildAnalysis, + TopicKeyIssueTrack, TopicKeyRecord, } @@ -117,3 +118,31 @@ func TestBuildAnalysisRoundTrip(t *testing.T) { assert.True(t, proto.Equal(analysis, got)) assert.JSONEq(t, `{"id":"build/42","queue_name":"validation/main"}`, string(data)) } + +func TestIssueTrackRoundTrip(t *testing.T) { + for _, tt := range []struct { + name string + finalize bool + wire string + }{ + {name: "observations", wire: `{"build_id":"build/42","queue_name":"validation/main"}`}, + {name: "final analysis", finalize: true, wire: `{"build_id":"build/42","queue_name":"validation/main","finalize":true}`}, + } { + t.Run(tt.name, func(t *testing.T) { + notification := &IssueTrack{BuildId: "build/42", QueueName: "validation/main", Finalize: tt.finalize} + data, err := Marshal(notification) + require.NoError(t, err) + assert.JSONEq(t, tt.wire, string(data)) + + got := &IssueTrack{} + require.NoError(t, Unmarshal(data, got)) + assert.True(t, proto.Equal(notification, got)) + }) + } +} + +func TestIssueTrackToleratesUnknownFields(t *testing.T) { + got := &IssueTrack{} + 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)) +} diff --git a/stovepipe/core/messagequeue/proto/BUILD.bazel b/stovepipe/core/messagequeue/proto/BUILD.bazel index 382e14f5b..e942b7f1f 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", + "issuetrack.proto", "process.proto", "record.proto", ], diff --git a/stovepipe/core/messagequeue/proto/issuetrack.proto b/stovepipe/core/messagequeue/proto/issuetrack.proto new file mode 100644 index 000000000..52acce9f5 --- /dev/null +++ b/stovepipe/core/messagequeue/proto/issuetrack.proto @@ -0,0 +1,39 @@ +// 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 = "IssueTrackProto"; +option java_package = "com.uber.submitqueue.stovepipe.messagequeue"; + +// IssueTrack requests queue-scoped reconciliation of persisted build observations. +// Publish only after successful detection and observation persistence; partition by queue. +message IssueTrack { + option (uber.base.messagequeue.topic_keys) = "issuetrack"; + + // build_id is the required opaque identifier of the persisted build. + string build_id = 1; + // queue_name is the required queue namespace of the build. + string queue_name = 2; + // finalize requests final reconciliation after successful terminal detection + // and persistence, including when no failures were detected. + // Otherwise the notification permits opening issues, not resolution by absence. + bool finalize = 3; +} diff --git a/stovepipe/core/messagequeue/protopb/BUILD.bazel b/stovepipe/core/messagequeue/protopb/BUILD.bazel index f1d2656fe..61c31ef92 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", + "issuetrack.pb.go", "process.pb.go", "record.pb.go", ], diff --git a/stovepipe/core/messagequeue/protopb/issuetrack.pb.go b/stovepipe/core/messagequeue/protopb/issuetrack.pb.go new file mode 100644 index 000000000..38a282c26 --- /dev/null +++ b/stovepipe/core/messagequeue/protopb/issuetrack.pb.go @@ -0,0 +1,167 @@ +// 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: issuetrack.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) +) + +// IssueTrack requests queue-scoped reconciliation of persisted build observations. +// Publish only after successful detection and observation persistence; partition by queue. +type IssueTrack struct { + state protoimpl.MessageState `protogen:"open.v1"` + // build_id is the required opaque identifier of the persisted build. + BuildId string `protobuf:"bytes,1,opt,name=build_id,json=buildId,proto3" json:"build_id,omitempty"` + // 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"` + // finalize requests final reconciliation after successful terminal detection + // and persistence, including when no failures were detected. + // Otherwise the notification permits opening issues, not resolution by absence. + Finalize bool `protobuf:"varint,3,opt,name=finalize,proto3" json:"finalize,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *IssueTrack) Reset() { + *x = IssueTrack{} + mi := &file_issuetrack_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *IssueTrack) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*IssueTrack) ProtoMessage() {} + +func (x *IssueTrack) ProtoReflect() protoreflect.Message { + mi := &file_issuetrack_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 IssueTrack.ProtoReflect.Descriptor instead. +func (*IssueTrack) Descriptor() ([]byte, []int) { + return file_issuetrack_proto_rawDescGZIP(), []int{0} +} + +func (x *IssueTrack) GetBuildId() string { + if x != nil { + return x.BuildId + } + return "" +} + +func (x *IssueTrack) GetQueueName() string { + if x != nil { + return x.QueueName + } + return "" +} + +func (x *IssueTrack) GetFinalize() bool { + if x != nil { + return x.Finalize + } + return false +} + +var File_issuetrack_proto protoreflect.FileDescriptor + +const file_issuetrack_proto_rawDesc = "" + + "\n" + + "\x10issuetrack.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"r\n" + + "\n" + + "IssueTrack\x12\x19\n" + + "\bbuild_id\x18\x01 \x01(\tR\abuildId\x12\x1d\n" + + "\n" + + "queue_name\x18\x02 \x01(\tR\tqueueName\x12\x1a\n" + + "\bfinalize\x18\x03 \x01(\bR\bfinalize:\x0e\x8a\xb5\x18\n" + + "issuetrackB\x81\x01\n" + + "+com.uber.submitqueue.stovepipe.messagequeueB\x0fIssueTrackProtoP\x01Z?github.com/uber/submitqueue/stovepipe/core/messagequeue/protopbb\x06proto3" + +var ( + file_issuetrack_proto_rawDescOnce sync.Once + file_issuetrack_proto_rawDescData []byte +) + +func file_issuetrack_proto_rawDescGZIP() []byte { + file_issuetrack_proto_rawDescOnce.Do(func() { + file_issuetrack_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_issuetrack_proto_rawDesc), len(file_issuetrack_proto_rawDesc))) + }) + return file_issuetrack_proto_rawDescData +} + +var file_issuetrack_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_issuetrack_proto_goTypes = []any{ + (*IssueTrack)(nil), // 0: uber.stovepipe.messagequeue.IssueTrack +} +var file_issuetrack_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_issuetrack_proto_init() } +func file_issuetrack_proto_init() { + if File_issuetrack_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_issuetrack_proto_rawDesc), len(file_issuetrack_proto_rawDesc)), + NumEnums: 0, + NumMessages: 1, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_issuetrack_proto_goTypes, + DependencyIndexes: file_issuetrack_proto_depIdxs, + MessageInfos: file_issuetrack_proto_msgTypes, + }.Build() + File_issuetrack_proto = out.File + file_issuetrack_proto_goTypes = nil + file_issuetrack_proto_depIdxs = nil +} diff --git a/stovepipe/core/messagequeue/topics.go b/stovepipe/core/messagequeue/topics.go index 09da535a6..433bd4fe9 100644 --- a/stovepipe/core/messagequeue/topics.go +++ b/stovepipe/core/messagequeue/topics.go @@ -45,6 +45,9 @@ const ( // TopicKeyBuildAnalysis carries build IDs for issue detection, partitioned by build ID. TopicKeyBuildAnalysis TopicKey = "buildanalysis" + // TopicKeyIssueTrack carries persisted observation notifications, partitioned by queue. + TopicKeyIssueTrack TopicKey = "issuetrack" + // 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/stovepipe/entity/build_issue.go b/stovepipe/entity/build_issue.go index 9d32a8e00..44fca4cc0 100644 --- a/stovepipe/entity/build_issue.go +++ b/stovepipe/entity/build_issue.go @@ -16,7 +16,7 @@ package entity // BuildIssue identifies a validation failure by task and optional unit. type BuildIssue struct { - // TaskID identifies the logical validation task, stable across builds and retries. + // TaskID identifies the logical validation task, stable across builds and retries; it must be nonempty. TaskID string `json:"task_id"` // UnitID identifies the failed unit within the task; empty for a task-wide issue. // Repeated observations of the same condition use the same TaskID and UnitID. diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index eaf20aee0..18b191e5a 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -28,7 +28,9 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte ## Build issue observations -`BuildIssueStore` accumulates immutable observations identified by build, task, and unit within the bound queue. The build ID is supplied separately from the detected issue. Repeated observations return `ErrAlreadyExists`, which callers may treat as successful persistence. Empty unit IDs represent task-wide observations. +`BuildIssueStore` accumulates immutable observations identified by build, task, and unit within the bound queue. The build ID is supplied separately from the detected issue. Repeated observations return `ErrAlreadyExists`, which callers may treat as successful persistence. Build and task IDs must be nonempty; empty unit IDs represent task-wide observations. + +Listing fixes the queue/build primary-key prefix and returns bounded pages in ascending bytewise task/unit order. A task/unit cursor continues after the last returned key; the zero cursor starts the range, and a unit without a task is invalid. Pages are not a snapshot across calls: concurrent observations can affect membership. Empty pages represent no observations in that range, not build success or completed analysis. Each observation is stored separately, not in a serialized collection. Observations are historical evidence; they do not track whether an issue is currently open or resolved. No update, deletion, or absence-based recovery is implied by this store. diff --git a/stovepipe/extension/storage/build_issue_store.go b/stovepipe/extension/storage/build_issue_store.go index 8633bc3c9..2bf799eb0 100644 --- a/stovepipe/extension/storage/build_issue_store.go +++ b/stovepipe/extension/storage/build_issue_store.go @@ -22,8 +22,25 @@ import ( "github.com/uber/submitqueue/stovepipe/entity" ) +// BuildIssueCursor is an exclusive position in ascending bytewise task/unit order. +type BuildIssueCursor struct { + // TaskID is the task component; empty with an empty UnitID starts the range. + TaskID string + // UnitID is the unit component; empty identifies a task-wide observation. + UnitID string +} + +// BuildIssueRange bounds a primary-key scan within one build. +type BuildIssueRange struct { + // After excludes this key and earlier keys; the zero cursor starts the range. + After BuildIssueCursor + // Limit is the positive maximum number of observations to return. + Limit int +} + // BuildIssueStore retains immutable observations keyed by build, task, and unit // within the bound queue. Presence records an observation, not a currently open issue. +// Observation keys require nonempty build and task IDs; an empty unit is task-wide. type BuildIssueStore interface { // Create returns ErrAlreadyExists for a previously recorded build/task/unit. // Callers may treat that duplicate as success when accumulating observations. @@ -31,4 +48,9 @@ type BuildIssueStore interface { // Get returns ErrNotFound when the build/task/unit has not been recorded. Get(ctx context.Context, buildID, taskID, unitID string) (entity.BuildIssue, error) + + // List returns at most Limit observations in ascending bytewise (task, unit) order. + // An empty slice ends the range; invalid bounds or an empty build ID return an error. + // Pages are not a snapshot: concurrent observations may change membership. + List(ctx context.Context, buildID string, bounds BuildIssueRange) ([]entity.BuildIssue, error) } diff --git a/stovepipe/extension/storage/mock/build_issue_store_mock.go b/stovepipe/extension/storage/mock/build_issue_store_mock.go index e97faa051..d7b6f917d 100644 --- a/stovepipe/extension/storage/mock/build_issue_store_mock.go +++ b/stovepipe/extension/storage/mock/build_issue_store_mock.go @@ -14,6 +14,7 @@ import ( reflect "reflect" entity "github.com/uber/submitqueue/stovepipe/entity" + storage "github.com/uber/submitqueue/stovepipe/extension/storage" gomock "go.uber.org/mock/gomock" ) @@ -69,3 +70,18 @@ func (mr *MockBuildIssueStoreMockRecorder) Get(ctx, buildID, taskID, unitID any) mr.mock.ctrl.T.Helper() return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockBuildIssueStore)(nil).Get), ctx, buildID, taskID, unitID) } + +// List mocks base method. +func (m *MockBuildIssueStore) List(ctx context.Context, buildID string, bounds storage.BuildIssueRange) ([]entity.BuildIssue, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "List", ctx, buildID, bounds) + ret0, _ := ret[0].([]entity.BuildIssue) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// List indicates an expected call of List. +func (mr *MockBuildIssueStoreMockRecorder) List(ctx, buildID, bounds any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockBuildIssueStore)(nil).List), ctx, buildID, bounds) +} diff --git a/stovepipe/extension/storage/mysql/BUILD.bazel b/stovepipe/extension/storage/mysql/BUILD.bazel index 2a448e178..1f3b25ea2 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -33,6 +33,7 @@ go_library( go_test( name = "go_default_test", srcs = [ + "build_issue_list_test.go", "build_issue_store_test.go", "build_store_test.go", "issue_log_store_test.go", diff --git a/stovepipe/extension/storage/mysql/build_issue_list_test.go b/stovepipe/extension/storage/mysql/build_issue_list_test.go new file mode 100644 index 000000000..e02471048 --- /dev/null +++ b/stovepipe/extension/storage/mysql/build_issue_list_test.go @@ -0,0 +1,138 @@ +// 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 mysql + +import ( + "context" + "database/sql/driver" + "regexp" + "strings" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func TestBuildIssueList(t *testing.T) { + for _, tt := range []struct { + name string + after storage.BuildIssueCursor + empty bool + dbErr error + scanError bool + rowError bool + }{ + {name: "first page"}, + {name: "after task-wide issue", after: storage.BuildIssueCursor{TaskID: "task"}}, + {name: "empty page", empty: true}, + {name: "query error", dbErr: assert.AnError}, + {name: "cancelled", dbErr: context.Canceled}, + {name: "scan error", scanError: true}, + {name: "iteration error", rowError: true}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + store := NewBuildIssueStore(db, testMetrics(), "queue") + issue := entity.BuildIssue{TaskID: "task", UnitID: "unit"} + query := listBuildIssuesQuery + args := []driver.Value{"queue", "build/1", int64(2)} + if tt.after.TaskID != "" { + query = listBuildIssuesAfterCursorQuery + args = []driver.Value{"queue", "build/1", tt.after.TaskID, tt.after.TaskID, tt.after.UnitID, int64(2)} + } + expectation := mock.ExpectQuery(regexp.QuoteMeta(query)).WithArgs(args...) + if tt.dbErr != nil { + expectation.WillReturnError(tt.dbErr) + } else { + rows := sqlmock.NewRows([]string{"task_id", "unit_id"}) + if !tt.empty { + var task any = issue.TaskID + if tt.scanError { + task = nil + } + rows.AddRow(task, issue.UnitID) + if tt.rowError { + rows.AddRow("task", "unit-2").RowError(1, assert.AnError) + } + } + expectation.WillReturnRows(rows).RowsWillBeClosed() + } + + got, err := store.List(context.Background(), "build/1", storage.BuildIssueRange{After: tt.after, Limit: 2}) + switch { + case tt.dbErr != nil: + require.ErrorIs(t, err, tt.dbErr) + require.Nil(t, got) + case tt.scanError: + require.Error(t, err) + require.Nil(t, got) + case tt.rowError: + require.ErrorIs(t, err, assert.AnError) + require.Nil(t, got) + case tt.empty: + require.NoError(t, err) + require.NotNil(t, got) + require.Empty(t, got) + default: + require.NoError(t, err) + require.Equal(t, []entity.BuildIssue{issue}, got) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestBuildIssueListRejectsInvalidRange(t *testing.T) { + for _, tt := range []struct { + name string + queue string + buildID string + bounds storage.BuildIssueRange + }{ + {name: "empty build", buildID: "", bounds: storage.BuildIssueRange{Limit: 1}}, + {name: "oversized build", buildID: strings.Repeat("b", 1021), bounds: storage.BuildIssueRange{Limit: 1}}, + {name: "zero limit", buildID: "build/1", bounds: storage.BuildIssueRange{}}, + {name: "negative limit", buildID: "build/1", bounds: storage.BuildIssueRange{Limit: -1}}, + {name: "unit without task", buildID: "build/1", bounds: storage.BuildIssueRange{Limit: 1, + After: storage.BuildIssueCursor{UnitID: "unit"}}}, + {name: "oversized task", buildID: "build/1", bounds: storage.BuildIssueRange{Limit: 1, + After: storage.BuildIssueCursor{TaskID: strings.Repeat("t", 256)}}}, + {name: "oversized unit", buildID: "build/1", bounds: storage.BuildIssueRange{Limit: 1, + After: storage.BuildIssueCursor{TaskID: "task", UnitID: strings.Repeat("😀", 257)}}}, + {name: "oversized queue", buildID: "build/1", queue: strings.Repeat("q", 256), bounds: storage.BuildIssueRange{Limit: 1}}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + queue := tt.queue + if queue == "" { + queue = "queue" + } + buildID := tt.buildID + store := NewBuildIssueStore(db, testMetrics(), queue) + mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id"})) + + got, err := store.List(context.Background(), buildID, tt.bounds) + require.Error(t, err) + require.Nil(t, got) + }) + } +} diff --git a/stovepipe/extension/storage/mysql/build_issue_store.go b/stovepipe/extension/storage/mysql/build_issue_store.go index a7c825b38..88755ced0 100644 --- a/stovepipe/extension/storage/mysql/build_issue_store.go +++ b/stovepipe/extension/storage/mysql/build_issue_store.go @@ -27,6 +27,17 @@ import ( "github.com/uber/submitqueue/stovepipe/extension/storage" ) +const listBuildIssuesQuery = ` + SELECT task_id, unit_id + FROM build_issue WHERE queue = ? AND build_id = ? + ORDER BY task_id ASC, unit_id ASC LIMIT ?` + +const listBuildIssuesAfterCursorQuery = ` + SELECT task_id, unit_id + FROM build_issue WHERE queue = ? AND build_id = ? + AND (task_id > ? OR (task_id = ? AND unit_id > ?)) + ORDER BY task_id ASC, unit_id ASC LIMIT ?` + type buildIssueStore struct { db *sql.DB scope tally.Scope @@ -42,7 +53,7 @@ func (s *buildIssueStore) Create(ctx context.Context, buildID string, issue enti op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) defer func() { op.Complete(retErr) }() - if err := validateBuildIssueKey(s.queue, buildID, issue.TaskID, issue.UnitID); err != nil { + if err := s.validateKey(buildID, issue.TaskID, issue.UnitID); err != nil { return err } _, err := s.db.ExecContext(ctx, @@ -65,7 +76,7 @@ func (s *buildIssueStore) Get(ctx context.Context, buildID, taskID, unitID strin op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) defer func() { op.Complete(retErr) }() - if err := validateBuildIssueKey(s.queue, buildID, taskID, unitID); err != nil { + if err := s.validateKey(buildID, taskID, unitID); err != nil { return entity.BuildIssue{}, err } @@ -86,6 +97,58 @@ func (s *buildIssueStore) Get(ctx context.Context, buildID, taskID, unitID strin return issue, nil } +func (s *buildIssueStore) List(ctx context.Context, buildID string, bounds storage.BuildIssueRange) (ret []entity.BuildIssue, retErr error) { + op := metrics.Begin(s.scope, "list", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if buildID == "" || bounds.Limit <= 0 { + return nil, fmt.Errorf("build issue range requires a build ID and positive limit") + } + cursor := bounds.After + if cursor.TaskID == "" && cursor.UnitID != "" { + return nil, fmt.Errorf("build issue cursor requires a task ID when a unit ID is supplied") + } + if err := validateBuildIssueKey(s.queue, buildID, cursor.TaskID, cursor.UnitID); err != nil { + return nil, err + } + + var ( + rows *sql.Rows + err error + ) + if cursor.TaskID == "" { + rows, err = s.db.QueryContext(ctx, listBuildIssuesQuery, s.queue, buildID, bounds.Limit) + } else { + rows, err = s.db.QueryContext(ctx, listBuildIssuesAfterCursorQuery, + s.queue, buildID, cursor.TaskID, cursor.TaskID, cursor.UnitID, bounds.Limit) + } + if err != nil { + return nil, fmt.Errorf("failed to list build issues: %w", err) + } + defer rows.Close() + + issues := make([]entity.BuildIssue, 0) + for rows.Next() { + var issue entity.BuildIssue + if err := rows.Scan(&issue.TaskID, &issue.UnitID); err != nil { + return nil, fmt.Errorf("failed to scan build issue: %w", err) + } + issues = append(issues, issue) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to iterate build issues: %w", err) + } + + return issues, nil +} + +func (s *buildIssueStore) validateKey(buildID, taskID, unitID string) error { + if buildID == "" || taskID == "" { + return fmt.Errorf("build issue requires nonempty build and task IDs") + } + return validateBuildIssueKey(s.queue, buildID, taskID, unitID) +} + func validateBuildIssueKey(queue, buildID, taskID, unitID string) error { for _, field := range []struct { name string diff --git a/stovepipe/extension/storage/mysql/build_issue_store_test.go b/stovepipe/extension/storage/mysql/build_issue_store_test.go index 47eeddc6f..c14054a4a 100644 --- a/stovepipe/extension/storage/mysql/build_issue_store_test.go +++ b/stovepipe/extension/storage/mysql/build_issue_store_test.go @@ -162,3 +162,28 @@ func TestBuildIssueStoreRejectsOversizedIdentifiers(t *testing.T) { }) } } + +func TestBuildIssueStoreRejectsMissingIdentifiers(t *testing.T) { + for _, tt := range []struct { + name string + buildID string + taskID string + }{ + {name: "missing build", taskID: "task"}, + {name: "missing task", buildID: "build/1"}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + store := NewBuildIssueStore(db, testMetrics(), "queue") + mock.MatchExpectationsInOrder(false) + mock.ExpectExec("INSERT INTO build_issue").WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id"}).AddRow(tt.taskID, "unit")) + + require.Error(t, store.Create(context.Background(), tt.buildID, entity.BuildIssue{TaskID: tt.taskID, UnitID: "unit"})) + _, err = store.Get(context.Background(), tt.buildID, tt.taskID, "unit") + require.Error(t, err) + }) + } +} diff --git a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index 726524bb9..0ba8776b6 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_test") go_test( name = "go_default_test", srcs = [ + "build_issue_list_test.go", "build_issue_test.go", "issue_log_test.go", "open_issue_list_test.go", diff --git a/test/integration/stovepipe/extension/storage/mysql/build_issue_list_test.go b/test/integration/stovepipe/extension/storage/mysql/build_issue_list_test.go new file mode 100644 index 000000000..ee1688ff8 --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/build_issue_list_test.go @@ -0,0 +1,156 @@ +// 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 mysql + +import ( + "context" + "fmt" + "sort" + "strings" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func testBuildIssueList(t *testing.T, ctx context.Context, factory storage.Factory) { + t.Helper() + + bound, err := factory.For(storage.Config{QueueName: "build-issue-list/main"}) + require.NoError(t, err) + store := bound.GetBuildIssueStore() + empty, err := store.List(ctx, "build/1", storage.BuildIssueRange{Limit: 2}) + require.NoError(t, err) + require.Empty(t, empty) + + expected := []entity.BuildIssue{ + {TaskID: "A"}, + {TaskID: "a"}, + {TaskID: "a", UnitID: "\x00"}, + {TaskID: "a", UnitID: "Unit"}, + {TaskID: "a", UnitID: "unit"}, + {TaskID: "a", UnitID: "unit "}, + {TaskID: "a", UnitID: "unit\x00"}, + {TaskID: "a ", UnitID: "unit"}, + {TaskID: "a\x00b", UnitID: "c"}, + {TaskID: "a", UnitID: "b\x00c"}, + {TaskID: "é", UnitID: "unit"}, + {TaskID: "😀", UnitID: "unit"}, + } + for i := len(expected) - 1; i >= 0; i-- { + require.NoError(t, store.Create(ctx, "build/1", expected[i])) + } + sort.Slice(expected, func(i, j int) bool { + if expected[i].TaskID != expected[j].TaskID { + return expected[i].TaskID < expected[j].TaskID + } + return expected[i].UnitID < expected[j].UnitID + }) + + t.Run("queue and build isolation", func(t *testing.T) { + otherQueue, err := factory.For(storage.Config{QueueName: "Build-issue-list/main"}) + require.NoError(t, err) + otherStore := otherQueue.GetBuildIssueStore() + empty, err := otherStore.List(ctx, "build/1", storage.BuildIssueRange{Limit: 2}) + require.NoError(t, err) + require.Empty(t, empty) + require.NoError(t, otherStore.Create(ctx, "build/1", expected[0])) + got, err := otherStore.List(ctx, "build/1", storage.BuildIssueRange{Limit: 2}) + require.NoError(t, err) + require.Equal(t, expected[:1], got) + + for _, buildID := range []string{"build/2", "Build/1", "build/1 "} { + require.NoError(t, store.Create(ctx, buildID, expected[0])) + got, err := store.List(ctx, buildID, storage.BuildIssueRange{Limit: 2}) + require.NoError(t, err) + require.Equal(t, expected[:1], got) + } + }) + + for _, limit := range []int{1, 2, 3, len(expected), len(expected) + 1} { + t.Run(fmt.Sprintf("page size %d", limit), func(t *testing.T) { + bounds := storage.BuildIssueRange{Limit: limit} + var got []entity.BuildIssue + for { + page, err := store.List(ctx, "build/1", bounds) + require.NoError(t, err) + require.LessOrEqual(t, len(page), limit) + if len(page) == 0 { + break + } + got = append(got, page...) + require.LessOrEqual(t, len(got), len(expected)) + last := page[len(page)-1] + bounds.After = storage.BuildIssueCursor{TaskID: last.TaskID, UnitID: last.UnitID} + } + require.Equal(t, expected, got) + }) + } + + t.Run("cursor need not identify an observation", func(t *testing.T) { + cursor := storage.BuildIssueCursor{TaskID: "a", UnitID: "missing"} + var want []entity.BuildIssue + for _, issue := range expected { + if issue.TaskID > cursor.TaskID || (issue.TaskID == cursor.TaskID && issue.UnitID > cursor.UnitID) { + want = append(want, issue) + } + } + got, err := store.List(ctx, "build/1", storage.BuildIssueRange{After: cursor, Limit: len(expected)}) + require.NoError(t, err) + require.Equal(t, want, got) + }) + + t.Run("identifier boundaries", func(t *testing.T) { + bound, err := factory.For(storage.Config{QueueName: strings.Repeat("q", 255)}) + require.NoError(t, err) + maxStore := bound.GetBuildIssueStore() + buildID := strings.Repeat("😀", 255) + issue := entity.BuildIssue{TaskID: strings.Repeat("t", 255), UnitID: strings.Repeat("😀", 256)} + require.NoError(t, maxStore.Create(ctx, buildID, issue)) + got, err := maxStore.List(ctx, buildID, storage.BuildIssueRange{Limit: 1}) + require.NoError(t, err) + require.Equal(t, []entity.BuildIssue{issue}, got) + + bounds := storage.BuildIssueRange{Limit: 1, + After: storage.BuildIssueCursor{TaskID: issue.TaskID, UnitID: issue.UnitID}} + got, err = maxStore.List(ctx, buildID, bounds) + require.NoError(t, err) + require.Empty(t, got) + _, err = maxStore.List(ctx, buildID+"x", bounds) + require.Error(t, err) + oversized := bounds + oversized.After.TaskID += "x" + _, err = maxStore.List(ctx, buildID, oversized) + require.Error(t, err) + oversized = bounds + oversized.After.UnitID += "x" + _, err = maxStore.List(ctx, buildID, oversized) + require.Error(t, err) + }) + + t.Run("missing identifiers are rejected", func(t *testing.T) { + for _, tt := range []struct{ buildID, taskID string }{ + {taskID: "task"}, {buildID: "build/invalid"}, + } { + issue := entity.BuildIssue{TaskID: tt.taskID, UnitID: "unit"} + require.Error(t, store.Create(ctx, tt.buildID, issue)) + _, err := store.Get(ctx, tt.buildID, tt.taskID, "unit") + require.Error(t, err) + } + _, err := store.List(ctx, "", storage.BuildIssueRange{Limit: 1}) + require.Error(t, err) + }) +} diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index b299f2982..e3b5f97e1 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -95,6 +95,11 @@ func TestMySQLStorage(t *testing.T) { testBuildIssueStore(t, ctx, db, factory) }) + t.Run("BuildIssueList", func(t *testing.T) { + resetStorage(t, db) + testBuildIssueList(t, ctx, factory) + }) + t.Run("OpenIssueStore", func(t *testing.T) { resetStorage(t, db) testOpenIssueStore(t, ctx, factory) diff --git a/tool/proto/BUILD.bazel b/tool/proto/BUILD.bazel index 62be9e389..8747096b1 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:issuetrack.proto", "//stovepipe/core/messagequeue/proto:process.proto", "//stovepipe/core/messagequeue/proto:record.proto", ],