From 8628279644e15c49abc09bf10df9e7b6f8076725 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Thu, 29 Oct 2026 20:10:51 +0000 Subject: [PATCH] feat(stovepipe): Add open issue listing ## Summary **What**: - Add bounded listing of current open issues within a queue. **Why**: - Establish the read behavior needed by the upcoming open-issues API. Storage already lists open-issue rows, but those rows can temporarily refer to closed issues or an earlier opening; callers must check the summaries before returning results. - Keep each read bounded while allowing callers to reach later issues. Advance continuation tokens past examined rows, including skipped ones, so stale entries cannot prevent pagination from making progress. ## Test Plan - [x] Add unit tests. ## Revert Plan - Revert this PR. Listing is not exposed through the service yet, and no persisted data changes. ## Issue Links - [CODEM-607](https://linear.app/uber/issue/CODEM-607/build-analysis) --- stovepipe/controller/BUILD.bazel | 2 + stovepipe/controller/list_open_issues.go | 200 +++++++++ stovepipe/controller/list_open_issues_test.go | 410 ++++++++++++++++++ stovepipe/controller/mock/BUILD.bazel | 1 + .../controller/mock/list_open_issues_mock.go | 57 +++ stovepipe/entity/BUILD.bazel | 1 + stovepipe/entity/list_open_issues.go | 34 ++ 7 files changed, 705 insertions(+) create mode 100644 stovepipe/controller/list_open_issues.go create mode 100644 stovepipe/controller/list_open_issues_test.go create mode 100644 stovepipe/controller/mock/list_open_issues_mock.go create mode 100644 stovepipe/entity/list_open_issues.go diff --git a/stovepipe/controller/BUILD.bazel b/stovepipe/controller/BUILD.bazel index 701635829..cd6cbcb87 100644 --- a/stovepipe/controller/BUILD.bazel +++ b/stovepipe/controller/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "get_queue_status.go", "ingest.go", "list.go", + "list_open_issues.go", "list_pagination.go", "ping.go", "project_status.go", @@ -40,6 +41,7 @@ go_test( "get_queue_status_test.go", "ingest_test.go", "list_consistency_test.go", + "list_open_issues_test.go", "list_pagination_test.go", "list_test.go", "ping_test.go", diff --git a/stovepipe/controller/list_open_issues.go b/stovepipe/controller/list_open_issues.go new file mode 100644 index 000000000..66b3a5499 --- /dev/null +++ b/stovepipe/controller/list_open_issues.go @@ -0,0 +1,200 @@ +// 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 controller + +//go:generate mockgen -source=list_open_issues.go -destination=mock/list_open_issues_mock.go -package=mock + +import ( + "context" + "encoding/base64" + "encoding/json" + "fmt" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/zap" +) + +const ( + openIssuesPageTokenVersion = 1 + maxOpenIssueCandidates = 500 +) + +type openIssuesPageToken struct { + Version int `json:"version"` + Queue string `json:"queue"` + LastTaskID string `json:"last_task_id"` + LastUnitID string `json:"last_unit_id"` +} + +// ListOpenIssuesController lists current open issues within a configured queue. +type ListOpenIssuesController interface { + ListOpenIssues(context.Context, entity.ListOpenIssuesRequest) (entity.ListOpenIssuesResult, error) +} + +type listOpenIssuesController struct { + logger *zap.SugaredLogger + scope tally.Scope + stores storage.Factory + configuredQueues map[string]struct{} +} + +// NewListOpenIssuesController restricts open-issue listing to configured queues. +func NewListOpenIssuesController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, configuredQueues []string) ListOpenIssuesController { + queues := make(map[string]struct{}, len(configuredQueues)) + for _, queue := range configuredQueues { + queues[queue] = struct{}{} + } + return &listOpenIssuesController{ + logger: logger, + scope: scope.SubScope("list_open_issues_controller"), + stores: stores, + configuredQueues: queues, + } +} + +func (c *listOpenIssuesController) ListOpenIssues(ctx context.Context, req entity.ListOpenIssuesRequest) (result entity.ListOpenIssuesResult, retErr error) { + op := metrics.Begin(c.scope, "list", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + + if req.Queue == "" { + return result, fmt.Errorf("ListOpenIssues requires a queue: %w", ErrInvalidRequest) + } + if _, ok := c.configuredQueues[req.Queue]; !ok { + return result, fmt.Errorf("ListOpenIssues queue %q is not configured: %w", req.Queue, ErrInvalidRequest) + } + + bounds, err := resolveOpenIssueRange(req) + if err != nil { + return result, err + } + + stores, err := c.stores.For(storage.Config{QueueName: req.Queue}) + if err != nil { + return result, fmt.Errorf("resolve open issue storage for queue %q: %w", req.Queue, err) + } + + scan, err := scanOpenIssuesPage(ctx, stores, req.Queue, bounds) + if err != nil { + return result, err + } + result.Issues = scan.issues + if scan.hasMore { + token, err := encodeOpenIssuesPageToken(openIssuesPageToken{ + Version: openIssuesPageTokenVersion, Queue: req.Queue, LastTaskID: scan.last.TaskID, LastUnitID: scan.last.UnitID, + }) + if err != nil { + return entity.ListOpenIssuesResult{}, fmt.Errorf("encode open issue continuation: %w", err) + } + result.NextPageToken = token + } + + c.logger.Debugw("queue open issues listed", "queue", req.Queue, "issue_count", len(result.Issues), "has_next_page", result.NextPageToken != "") + return result, nil +} + +type openIssuesPageScan struct { + issues []entity.IssueSummary + last entity.OpenIssue // Last inspected entry, even if skipped. + hasMore bool +} + +func scanOpenIssuesPage(ctx context.Context, stores storage.Storage, queue string, bounds storage.OpenIssueRange) (openIssuesPageScan, error) { + pageSize := bounds.Limit - 1 + scan := openIssuesPageScan{issues: make([]entity.IssueSummary, 0, pageSize)} + openIssues := stores.GetOpenIssueStore() + summaries := stores.GetIssueSummaryStore() + inspected := 0 + for { + want := min(pageSize-len(scan.issues), maxOpenIssueCandidates-inspected) + bounds.Limit = want + 1 + batch, err := openIssues.List(ctx, bounds) + if err != nil { + return openIssuesPageScan{}, fmt.Errorf("list open issues for queue %q: %w", queue, err) + } + for _, openIssue := range batch[:min(len(batch), want)] { + summary, err := summaries.Get(ctx, openIssue.TaskID, openIssue.UnitID) + if err != nil { + return openIssuesPageScan{}, fmt.Errorf("read open issue summary queue=%q task=%q unit=%q: %w", queue, openIssue.TaskID, openIssue.UnitID, err) + } + if summary.TaskID != openIssue.TaskID || summary.UnitID != openIssue.UnitID { + return openIssuesPageScan{}, fmt.Errorf("open issue disagrees with summary identity queue=%q task=%q unit=%q", queue, openIssue.TaskID, openIssue.UnitID) + } + if summary.Status == entity.IssueStatusOpen && summary.FirstSeenBuildID == openIssue.FirstSeenBuildID { + scan.issues = append(scan.issues, summary) + } + scan.last = openIssue + inspected++ + } + if len(batch) <= want { + return scan, nil + } + if len(scan.issues) == pageSize || inspected == maxOpenIssueCandidates { + scan.hasMore = true + return scan, nil + } + bounds.After = storage.OpenIssueCursor{TaskID: scan.last.TaskID, UnitID: scan.last.UnitID} + } +} + +func resolveOpenIssueRange(req entity.ListOpenIssuesRequest) (storage.OpenIssueRange, error) { + if req.PageSize < 0 || req.PageSize > maxListPageSize { + return storage.OpenIssueRange{}, fmt.Errorf("ListOpenIssues page_size must be between 0 and %d: %w", maxListPageSize, ErrInvalidRequest) + } + pageSize := int(req.PageSize) + if pageSize == 0 { + pageSize = defaultListPageSize + } + bounds := storage.OpenIssueRange{Limit: pageSize + 1} + if req.PageToken != "" { + token, err := decodeOpenIssuesPageToken(req.PageToken) + if err != nil { + return storage.OpenIssueRange{}, err + } + if token.Queue != req.Queue { + return storage.OpenIssueRange{}, fmt.Errorf("ListOpenIssues page_token does not match the queue: %w", ErrInvalidRequest) + } + bounds.After = storage.OpenIssueCursor{TaskID: token.LastTaskID, UnitID: token.LastUnitID} + } + return bounds, nil +} + +func encodeOpenIssuesPageToken(token openIssuesPageToken) (string, error) { + contents, err := json.Marshal(token) + if err != nil { + return "", err + } + return base64.RawURLEncoding.EncodeToString(contents), nil +} + +func decodeOpenIssuesPageToken(encoded string) (openIssuesPageToken, error) { + contents, err := base64.RawURLEncoding.Strict().DecodeString(encoded) + if err != nil { + return openIssuesPageToken{}, fmt.Errorf("ListOpenIssues invalid page_token encoding: %w", ErrInvalidRequest) + } + var token openIssuesPageToken + if err := json.Unmarshal(contents, &token); err != nil { + return openIssuesPageToken{}, fmt.Errorf("ListOpenIssues invalid page_token contents: %w", ErrInvalidRequest) + } + if token.Version != openIssuesPageTokenVersion || token.Queue == "" || token.LastTaskID == "" { + return openIssuesPageToken{}, fmt.Errorf("ListOpenIssues invalid page_token fields: %w", ErrInvalidRequest) + } + if len(token.LastTaskID) > 255 || len(token.LastUnitID) > 1024 { + return openIssuesPageToken{}, fmt.Errorf("ListOpenIssues page_token key exceeds identifier limits: %w", ErrInvalidRequest) + } + return token, nil +} diff --git a/stovepipe/controller/list_open_issues_test.go b/stovepipe/controller/list_open_issues_test.go new file mode 100644 index 000000000..02e2e5c0e --- /dev/null +++ b/stovepipe/controller/list_open_issues_test.go @@ -0,0 +1,410 @@ +// 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 controller + +import ( + "context" + "encoding/base64" + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +const openIssueListQueue = "queue/main" + +type openIssueListFixture struct { + controller ListOpenIssuesController + factory *storagemock.MockFactory + stores *storagemock.MockStorage + openIssues *storagemock.MockOpenIssueStore + summaries *storagemock.MockIssueSummaryStore +} + +func newOpenIssueListFixture(t *testing.T) openIssueListFixture { + t.Helper() + ctrl := gomock.NewController(t) + f := openIssueListFixture{factory: storagemock.NewMockFactory(ctrl), stores: storagemock.NewMockStorage(ctrl), + openIssues: storagemock.NewMockOpenIssueStore(ctrl), summaries: storagemock.NewMockIssueSummaryStore(ctrl)} + f.controller = NewListOpenIssuesController(zap.NewNop().Sugar(), tally.NoopScope, f.factory, []string{openIssueListQueue}) + f.stores.EXPECT().GetOpenIssueStore().Return(f.openIssues).AnyTimes() + f.stores.EXPECT().GetIssueSummaryStore().Return(f.summaries).AnyTimes() + return f +} + +func (f openIssueListFixture) expectOpenIssues(bounds storage.OpenIssueRange, openIssues []entity.OpenIssue) { + f.factory.EXPECT().For(storage.Config{QueueName: openIssueListQueue}).Return(f.stores, nil) + f.openIssues.EXPECT().List(gomock.Any(), bounds).Return(openIssues, nil) +} + +func summaryForOpenIssue(openIssue entity.OpenIssue) entity.IssueSummary { + return entity.IssueSummary{TaskID: openIssue.TaskID, UnitID: openIssue.UnitID, Status: entity.IssueStatusOpen, + FirstSeenBuildID: openIssue.FirstSeenBuildID, RequestID: "100", BuildID: openIssue.FirstSeenBuildID, ChangedAtMs: 1000, Version: 1} +} + +func TestListOpenIssuesReturnsBoundedPages(t *testing.T) { + rows := []entity.OpenIssue{ + {TaskID: "a", FirstSeenBuildID: "build/1"}, + {TaskID: "a", UnitID: "unit", FirstSeenBuildID: "build/2"}, + {TaskID: "b", FirstSeenBuildID: "build/3"}, + } + for _, count := range []int{0, 1, 2, 3} { + t.Run(fmt.Sprintf("%d open issues", count), func(t *testing.T) { + f := newOpenIssueListFixture(t) + f.expectOpenIssues(storage.OpenIssueRange{Limit: 3}, rows[:count]) + want := make([]entity.IssueSummary, 0) + for _, row := range rows[:min(count, 2)] { + summary := summaryForOpenIssue(row) + want = append(want, summary) + f.summaries.EXPECT().Get(gomock.Any(), row.TaskID, row.UnitID).Return(summary, nil) + } + got, err := f.controller.ListOpenIssues(context.Background(), entity.ListOpenIssuesRequest{Queue: openIssueListQueue, PageSize: 2}) + require.NoError(t, err) + require.Equal(t, want, got.Issues) + if count <= 2 { + require.Empty(t, got.NextPageToken) + } else { + token, err := decodeOpenIssuesPageToken(got.NextPageToken) + require.NoError(t, err) + require.Equal(t, openIssuesPageToken{Version: 1, Queue: openIssueListQueue, LastTaskID: rows[1].TaskID, LastUnitID: rows[1].UnitID}, token) + } + }) + } +} + +func TestListOpenIssuesFillsPagesPastStaleEntries(t *testing.T) { + rows := []entity.OpenIssue{ + {TaskID: "a", FirstSeenBuildID: "build/1"}, + {TaskID: "a", UnitID: "unit", FirstSeenBuildID: "build/2"}, + {TaskID: "b", FirstSeenBuildID: "build/3"}, + {TaskID: "b", UnitID: "unit", FirstSeenBuildID: "build/4"}, + {TaskID: "c", FirstSeenBuildID: "build/5"}, + {TaskID: "d", FirstSeenBuildID: "build/6"}, + } + cursor := func(row entity.OpenIssue) storage.OpenIssueCursor { + return storage.OpenIssueCursor{TaskID: row.TaskID, UnitID: row.UnitID} + } + type read struct { + bounds storage.OpenIssueRange + rows []entity.OpenIssue + } + for _, tt := range []struct { + name string + pageSize int32 + reads []read + wantRows []entity.OpenIssue + wantLast storage.OpenIssueCursor + }{ + { + name: "fills the page after closed and reopened references", pageSize: 3, + reads: []read{ + {bounds: storage.OpenIssueRange{Limit: 4}, rows: rows[:4]}, + {bounds: storage.OpenIssueRange{After: cursor(rows[2]), Limit: 3}, rows: rows[3:]}, + }, + wantRows: []entity.OpenIssue{rows[0], rows[3], rows[4]}, wantLast: cursor(rows[4]), + }, + { + name: "continues across several batches without skipping lookahead", pageSize: 2, + reads: []read{ + {bounds: storage.OpenIssueRange{Limit: 3}, rows: rows[:3]}, + {bounds: storage.OpenIssueRange{After: cursor(rows[1]), Limit: 2}, rows: rows[2:4]}, + {bounds: storage.OpenIssueRange{After: cursor(rows[2]), Limit: 2}, rows: rows[3:5]}, + }, + wantRows: []entity.OpenIssue{rows[0], rows[3]}, wantLast: cursor(rows[3]), + }, + { + name: "returns a partial final page without continuation", pageSize: 3, + reads: []read{ + {bounds: storage.OpenIssueRange{Limit: 4}, rows: rows[:4]}, + {bounds: storage.OpenIssueRange{After: cursor(rows[2]), Limit: 3}, rows: rows[3:4]}, + }, + wantRows: []entity.OpenIssue{rows[0], rows[3]}, + }, + { + name: "returns a full final page without continuation", pageSize: 3, + reads: []read{ + {bounds: storage.OpenIssueRange{Limit: 4}, rows: rows[:4]}, + {bounds: storage.OpenIssueRange{After: cursor(rows[2]), Limit: 3}, rows: rows[3:5]}, + }, + wantRows: []entity.OpenIssue{rows[0], rows[3], rows[4]}, + }, + } { + t.Run(tt.name, func(t *testing.T) { + f := newOpenIssueListFixture(t) + f.factory.EXPECT().For(storage.Config{QueueName: openIssueListQueue}).Return(f.stores, nil) + var calls []any + for _, read := range tt.reads { + calls = append(calls, f.openIssues.EXPECT().List(gomock.Any(), read.bounds).Return(read.rows, nil)) + for _, row := range read.rows[:min(len(read.rows), read.bounds.Limit-1)] { + summary := summaryForOpenIssue(row) + switch row { + case rows[1]: + summary.Status = entity.IssueStatusClosed + case rows[2]: + summary.FirstSeenBuildID = "new-opening" + } + f.summaries.EXPECT().Get(gomock.Any(), row.TaskID, row.UnitID).Return(summary, nil) + } + } + gomock.InOrder(calls...) + + got, err := f.controller.ListOpenIssues(context.Background(), entity.ListOpenIssuesRequest{Queue: openIssueListQueue, PageSize: tt.pageSize}) + require.NoError(t, err) + want := make([]entity.IssueSummary, 0, len(tt.wantRows)) + for _, row := range tt.wantRows { + want = append(want, summaryForOpenIssue(row)) + } + require.Equal(t, want, got.Issues) + if tt.wantLast.TaskID == "" { + require.Empty(t, got.NextPageToken) + } else { + token, err := decodeOpenIssuesPageToken(got.NextPageToken) + require.NoError(t, err) + require.Equal(t, tt.wantLast, storage.OpenIssueCursor{TaskID: token.LastTaskID, UnitID: token.LastUnitID}) + } + }) + } +} + +func TestListOpenIssuesStopsAtInspectionBudget(t *testing.T) { + for _, tt := range []struct { + name string + total int + firstOpen bool + }{ + {name: "empty page with continuation", total: 501}, + {name: "partial page with continuation", total: 501, firstOpen: true}, + {name: "empty exhausted range", total: 500}, + {name: "partial exhausted range", total: 500, firstOpen: true}, + } { + t.Run(tt.name, func(t *testing.T) { + f := newOpenIssueListFixture(t) + rows := make([]entity.OpenIssue, tt.total) + for i := range rows { + rows[i] = entity.OpenIssue{TaskID: fmt.Sprintf("task-%03d", i+1), FirstSeenBuildID: "build/1"} + } + calls := 1 + if tt.total > 500 { + calls++ + } + f.factory.EXPECT().For(storage.Config{QueueName: openIssueListQueue}).Return(f.stores, nil).Times(calls) + inspected := 0 + f.openIssues.EXPECT().List(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, bounds storage.OpenIssueRange) ([]entity.OpenIssue, error) { + after := storage.OpenIssueCursor{} + if inspected > 0 { + after.TaskID = rows[inspected-1].TaskID + } + require.Equal(t, after, bounds.After) + require.Positive(t, bounds.Limit) + require.LessOrEqual(t, bounds.Limit, 51) + return rows[inspected:min(len(rows), inspected+bounds.Limit)], nil + }).AnyTimes() + f.summaries.EXPECT().Get(gomock.Any(), gomock.Any(), "").DoAndReturn( + func(_ context.Context, taskID, _ string) (entity.IssueSummary, error) { + row := rows[inspected] + require.Equal(t, row.TaskID, taskID) + summary := summaryForOpenIssue(row) + if inspected != 500 && !(tt.firstOpen && inspected == 0) { + summary.Status = entity.IssueStatusClosed + } + inspected++ + return summary, nil + }).Times(tt.total) + + got, err := f.controller.ListOpenIssues(context.Background(), entity.ListOpenIssuesRequest{Queue: openIssueListQueue, PageSize: 50}) + require.NoError(t, err) + require.Equal(t, 500, inspected) + if tt.firstOpen { + require.Equal(t, []entity.IssueSummary{summaryForOpenIssue(rows[0])}, got.Issues) + } else { + require.Empty(t, got.Issues) + } + if tt.total == 500 { + require.Empty(t, got.NextPageToken) + return + } + token, err := decodeOpenIssuesPageToken(got.NextPageToken) + require.NoError(t, err) + require.Equal(t, rows[499].TaskID, token.LastTaskID) + require.Empty(t, token.LastUnitID) + + next, err := f.controller.ListOpenIssues(context.Background(), entity.ListOpenIssuesRequest{ + Queue: openIssueListQueue, PageSize: 50, PageToken: got.NextPageToken, + }) + require.NoError(t, err) + require.Equal(t, []entity.IssueSummary{summaryForOpenIssue(rows[500])}, next.Issues) + require.Empty(t, next.NextPageToken) + }) + } +} + +func TestListOpenIssuesDiscardsPartialPageOnLaterReadFailure(t *testing.T) { + for _, point := range []string{"lookup", "summary"} { + t.Run(point, func(t *testing.T) { + f := newOpenIssueListFixture(t) + rows := []entity.OpenIssue{ + {TaskID: "a", FirstSeenBuildID: "build/1"}, + {TaskID: "b", FirstSeenBuildID: "build/1"}, + {TaskID: "c", FirstSeenBuildID: "build/1"}, + } + f.expectOpenIssues(storage.OpenIssueRange{Limit: 3}, rows) + f.summaries.EXPECT().Get(gomock.Any(), "a", "").Return(summaryForOpenIssue(rows[0]), nil) + closed := summaryForOpenIssue(rows[1]) + closed.Status = entity.IssueStatusClosed + f.summaries.EXPECT().Get(gomock.Any(), "b", "").Return(closed, nil) + read := f.openIssues.EXPECT().List(gomock.Any(), storage.OpenIssueRange{After: storage.OpenIssueCursor{TaskID: "b"}, Limit: 2}) + if point == "lookup" { + read.Return(nil, assert.AnError) + } else { + read.Return(rows[2:], nil) + f.summaries.EXPECT().Get(gomock.Any(), "c", "").Return(entity.IssueSummary{}, assert.AnError) + } + got, err := f.controller.ListOpenIssues(context.Background(), entity.ListOpenIssuesRequest{Queue: openIssueListQueue, PageSize: 2}) + require.ErrorIs(t, err, assert.AnError) + require.Empty(t, got) + }) + } +} + +func TestListOpenIssuesUsesCurrentSummaryOnContinuation(t *testing.T) { + f := newOpenIssueListFixture(t) + token, err := encodeOpenIssuesPageToken(openIssuesPageToken{Version: 1, Queue: openIssueListQueue, LastTaskID: "a"}) + require.NoError(t, err) + row := entity.OpenIssue{TaskID: "b", FirstSeenBuildID: "build/1"} + f.expectOpenIssues(storage.OpenIssueRange{After: storage.OpenIssueCursor{TaskID: "a"}, Limit: 2}, []entity.OpenIssue{row}) + nowClosed := summaryForOpenIssue(row) + nowClosed.Status = entity.IssueStatusClosed + f.summaries.EXPECT().Get(gomock.Any(), row.TaskID, row.UnitID).Return(nowClosed, nil) + got, err := f.controller.ListOpenIssues(context.Background(), entity.ListOpenIssuesRequest{Queue: openIssueListQueue, PageSize: 1, PageToken: token}) + require.NoError(t, err) + require.Empty(t, got.Issues) + require.Empty(t, got.NextPageToken) +} + +func TestListOpenIssuesReturnsStorageErrors(t *testing.T) { + for _, point := range []string{"factory", "open issues", "summary", "missing summary", "cancellation", "wrong summary identity"} { + t.Run(point, func(t *testing.T) { + f := newOpenIssueListFixture(t) + dependencyErr := assert.AnError + if point == "factory" { + f.factory.EXPECT().For(storage.Config{QueueName: openIssueListQueue}).Return(nil, dependencyErr) + } else { + f.factory.EXPECT().For(storage.Config{QueueName: openIssueListQueue}).Return(f.stores, nil) + if point == "open issues" { + f.openIssues.EXPECT().List(gomock.Any(), storage.OpenIssueRange{Limit: 2}).Return(nil, dependencyErr) + } else { + row := entity.OpenIssue{TaskID: "task", FirstSeenBuildID: "build/1"} + f.openIssues.EXPECT().List(gomock.Any(), storage.OpenIssueRange{Limit: 2}).Return([]entity.OpenIssue{row}, nil) + switch point { + case "missing summary": + dependencyErr = storage.ErrNotFound + case "cancellation": + dependencyErr = context.Canceled + } + if point == "wrong summary identity" { + wrong := summaryForOpenIssue(row) + wrong.UnitID = "other" + f.summaries.EXPECT().Get(gomock.Any(), row.TaskID, row.UnitID).Return(wrong, nil) + } else { + f.summaries.EXPECT().Get(gomock.Any(), row.TaskID, row.UnitID).Return(entity.IssueSummary{}, dependencyErr) + } + } + } + got, err := f.controller.ListOpenIssues(context.Background(), entity.ListOpenIssuesRequest{Queue: openIssueListQueue, PageSize: 1}) + require.Error(t, err) + if point != "wrong summary identity" { + require.ErrorIs(t, err, dependencyErr) + } + require.False(t, IsInvalidRequest(err)) + require.Empty(t, got) + }) + } +} + +func TestOpenIssuesTokenRoundTrip(t *testing.T) { + for _, unit := range []string{"", "unit ", "a\x00b", "😀"} { + want := openIssuesPageToken{Version: 1, Queue: openIssueListQueue, LastTaskID: "task", LastUnitID: unit} + encoded, err := encodeOpenIssuesPageToken(want) + require.NoError(t, err) + got, err := decodeOpenIssuesPageToken(encoded) + require.NoError(t, err) + require.Equal(t, want, got) + } +} + +func TestListOpenIssuesRejectsInvalidInputBeforeStorage(t *testing.T) { + valid := openIssuesPageToken{Version: 1, Queue: openIssueListQueue, LastTaskID: "task"} + for _, tt := range []struct { + name string + req entity.ListOpenIssuesRequest + tokenChange func(*openIssuesPageToken) + }{ + {name: "missing queue"}, + {name: "unconfigured queue", req: entity.ListOpenIssuesRequest{Queue: "other"}}, + {name: "negative size", req: entity.ListOpenIssuesRequest{Queue: openIssueListQueue, PageSize: -1}}, + {name: "oversized page", req: entity.ListOpenIssuesRequest{Queue: openIssueListQueue, PageSize: 201}}, + {name: "bad encoding", req: entity.ListOpenIssuesRequest{Queue: openIssueListQueue, PageToken: "%"}}, + {name: "bad JSON", req: entity.ListOpenIssuesRequest{Queue: openIssueListQueue, PageToken: base64.RawURLEncoding.EncodeToString([]byte("not json"))}}, + {name: "wrong version", tokenChange: func(p *openIssuesPageToken) { p.Version++ }}, + {name: "empty token queue", tokenChange: func(p *openIssuesPageToken) { p.Queue = "" }}, + {name: "different queue", tokenChange: func(p *openIssuesPageToken) { p.Queue = "other" }}, + {name: "missing task", tokenChange: func(p *openIssuesPageToken) { p.LastTaskID = "" }}, + {name: "oversized task", tokenChange: func(p *openIssuesPageToken) { p.LastTaskID = strings.Repeat("t", 256) }}, + {name: "oversized unit", tokenChange: func(p *openIssuesPageToken) { p.LastUnitID = strings.Repeat("u", 1025) }}, + } { + t.Run(tt.name, func(t *testing.T) { + req := tt.req + if tt.tokenChange != nil { + token := valid + tt.tokenChange(&token) + var err error + req.Queue = openIssueListQueue + req.PageToken, err = encodeOpenIssuesPageToken(token) + require.NoError(t, err) + } + f := newOpenIssueListFixture(t) + _, err := f.controller.ListOpenIssues(context.Background(), req) + require.True(t, IsInvalidRequest(err)) + }) + } +} + +func TestOpenIssuesRangeUsesBoundedDefaults(t *testing.T) { + for _, tt := range []struct { + size int32 + limit int + }{{0, 51}, {1, 2}, {200, 201}} { + bounds, err := resolveOpenIssueRange(entity.ListOpenIssuesRequest{Queue: openIssueListQueue, PageSize: tt.size}) + require.NoError(t, err) + require.Equal(t, tt.limit, bounds.Limit) + } +} + +func TestOpenIssuesRejectsRequestListTokens(t *testing.T) { + token, err := encodeListPageToken(listPageToken{Version: listPageTokenVersion, Queue: openIssueListQueue, + AcceptedBeforeMs: 1000, LastAcceptedAtMs: 500, LastRequestID: "100"}) + require.NoError(t, err) + _, err = resolveOpenIssueRange(entity.ListOpenIssuesRequest{Queue: openIssueListQueue, PageToken: token}) + require.True(t, IsInvalidRequest(err)) +} diff --git a/stovepipe/controller/mock/BUILD.bazel b/stovepipe/controller/mock/BUILD.bazel index d7f572103..32f9f1f51 100644 --- a/stovepipe/controller/mock/BUILD.bazel +++ b/stovepipe/controller/mock/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library") go_library( name = "go_default_library", srcs = [ + "list_open_issues_mock.go", "project_status_mock.go", "request_history_mock.go", ], diff --git a/stovepipe/controller/mock/list_open_issues_mock.go b/stovepipe/controller/mock/list_open_issues_mock.go new file mode 100644 index 000000000..9393fa550 --- /dev/null +++ b/stovepipe/controller/mock/list_open_issues_mock.go @@ -0,0 +1,57 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: list_open_issues.go +// +// Generated by this command: +// +// mockgen -source=list_open_issues.go -destination=mock/list_open_issues_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/stovepipe/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockListOpenIssuesController is a mock of ListOpenIssuesController interface. +type MockListOpenIssuesController struct { + ctrl *gomock.Controller + recorder *MockListOpenIssuesControllerMockRecorder + isgomock struct{} +} + +// MockListOpenIssuesControllerMockRecorder is the mock recorder for MockListOpenIssuesController. +type MockListOpenIssuesControllerMockRecorder struct { + mock *MockListOpenIssuesController +} + +// NewMockListOpenIssuesController creates a new mock instance. +func NewMockListOpenIssuesController(ctrl *gomock.Controller) *MockListOpenIssuesController { + mock := &MockListOpenIssuesController{ctrl: ctrl} + mock.recorder = &MockListOpenIssuesControllerMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockListOpenIssuesController) EXPECT() *MockListOpenIssuesControllerMockRecorder { + return m.recorder +} + +// ListOpenIssues mocks base method. +func (m *MockListOpenIssuesController) ListOpenIssues(arg0 context.Context, arg1 entity.ListOpenIssuesRequest) (entity.ListOpenIssuesResult, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ListOpenIssues", arg0, arg1) + ret0, _ := ret[0].(entity.ListOpenIssuesResult) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ListOpenIssues indicates an expected call of ListOpenIssues. +func (mr *MockListOpenIssuesControllerMockRecorder) ListOpenIssues(arg0, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListOpenIssues", reflect.TypeOf((*MockListOpenIssuesController)(nil).ListOpenIssues), arg0, arg1) +} diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index a8c5cd632..ea53fdf11 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -11,6 +11,7 @@ go_library( "issue_summary.go", "issue_tracking_state.go", "list.go", + "list_open_issues.go", "open_issue.go", "project_fact_by_degree.go", "project_status.go", diff --git a/stovepipe/entity/list_open_issues.go b/stovepipe/entity/list_open_issues.go new file mode 100644 index 000000000..aede9ddde --- /dev/null +++ b/stovepipe/entity/list_open_issues.go @@ -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. + +package entity + +// ListOpenIssuesRequest selects current open issues within one queue. +type ListOpenIssuesRequest struct { + // Queue is the required configured queue containing the issues. + Queue string + // PageSize bounds returned issues, from 1 to 200; zero selects 50. + PageSize int32 + // PageToken is an opaque continuation bound to the queue and task/unit order. + PageToken string +} + +// ListOpenIssuesResult is a bounded page of current open issue summaries. +type ListOpenIssuesResult struct { + // Issues contains matching open summaries in ascending bytewise task/unit order. + Issues []IssueSummary + // NextPageToken continues after the last examined open-issue row, including skipped entries. + // Empty means no further open-issue row was observed, not that the queue is healthy. + NextPageToken string +}