feat(stovepipe): List open issues in pages (#924)

## Summary

**What**:

- List currently open issues in bounded, cursor-based pages within each
validation queue, preserving byte-exact task and unit ordering.
- Reject open issues without a task identity while preserving task-wide
issues with no unit.

**Why**:

- Enable issue reconciliation to discover existing open issues through
primary-key reads without secondary indexes or loading every issue at
once.

## Test Plan

- [x] Add unit tests.
- [x] Verify page boundaries, queue isolation, identifier limits, and
continuation after deletion against MySQL.

## Revert Plan

- Revert this PR. It changes storage reads and input validation without
a schema migration or enabling runtime writers.

## Issue Links

- [CODEM-607](https://linear.app/uber/issue/CODEM-607/build-analysis)

## Stack

1. #923
1. #924
1. #925

Author: prathshenoy

stovepipe/entity/open_issue.go

@@ -18,7 +18,7 @@ package entity
 // Its task and unit identify the condition; the first-seen build identifies
 // this opening and is immutable until removal.
 type OpenIssue struct {
-	// TaskID identifies the logical validation task.
+	// TaskID identifies the logical validation task and must be nonempty.
 	TaskID string `json:"task_id"`
 	// UnitID identifies the validation unit; empty for a task-wide issue.
 	UnitID string `json:"unit_id"`

stovepipe/extension/storage/README.md

@@ -36,13 +36,17 @@ The MySQL primary key contains the queue, build ID, task ID, and unit ID directl
 
 ## Currently open issues
 
-`OpenIssueStore` contains only currently open issues, keyed by task and unit within the bound queue. Removing an open row never removes immutable per-build observations. There is no open/resolved flag; a resolved issue has no row here.
+`OpenIssueStore` contains only currently open issues, keyed by task and unit within the bound queue. Removing an open row never removes immutable per-build observations. Task IDs must be nonempty; an empty unit identifies a task-wide issue. There is no open/resolved flag; a resolved issue has no row here.
 
 An open row is immutable. Its first-seen build identifies the current opening, not its latest observation. Repeated observations remain in the per-build store. Reopening after removal uses a different first-seen build ID; no last-observed build, version counter, or update operation is stored here.
 
+Listing scans the bound queue's primary-key prefix in ascending bytewise task/unit order, returning at most the requested positive limit. Continue after the last returned task/unit key; the zero cursor starts the range. A cursor with a unit but no task is invalid. Empty units remain task-wide issues and can be used as continuation boundaries. No secondary index or schema change is required.
+
+Pages are not a snapshot across calls. Concurrent changes can affect which rows appear, and the cursor row need not still exist. Readers must not treat listing as an atomic replacement or resolution decision; guarded deletion still protects a later opening.
+
 Deletion matches the task, unit, and first-seen build within the bound queue. It is safe to repeat: an absent opening is a no-op, and a stale deletion never removes a different opening of the same condition. Database failures are returned for retry.
 
-MySQL identifiers use the same binary byte limits as per-build observations. Build IDs are opaque context, not ordering tokens. The store does not choose resolution, enumerate issues, write lifecycle history, or reconcile analysis responses. Lifecycle occurrences are retained separately by `IssueLogStore`. Source-order reconciliation belongs to later controllers, not this membership store.
+MySQL identifiers use the same binary byte limits as per-build observations. Build IDs are opaque context, not ordering tokens. The store does not choose resolution, write lifecycle history, or reconcile analysis responses. Lifecycle occurrences are retained separately by `IssueLogStore`. Source-order reconciliation belongs to later controllers, not this membership store.
 
 ## Issue lifecycle history
 

stovepipe/extension/storage/mock/open_issue_store_mock.go

@@ -36,6 +36,7 @@ go_test(
         "build_issue_store_test.go",
         "build_store_test.go",
         "issue_log_store_test.go",
+        "open_issue_list_test.go",
         "open_issue_store_test.go",
         "project_fact_by_degree_store_test.go",
         "queue_store_test.go",

stovepipe/extension/storage/mysql/BUILD.bazel

@@ -0,0 +1,134 @@
+// 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 TestOpenIssueList(t *testing.T) {
+	for _, tt := range []struct {
+		name      string
+		after     storage.OpenIssueCursor
+		empty     bool
+		dbErr     error
+		scanError bool
+		rowError  bool
+	}{
+		{name: "first page"},
+		{name: "after task-wide issue", after: storage.OpenIssueCursor{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 := NewOpenIssueStore(db, testMetrics(), "queue")
+			issue := openIssueFixture()
+			query := listOpenIssuesQuery
+			args := []driver.Value{"queue", int64(2)}
+			if tt.after.TaskID != "" {
+				query = listOpenIssuesAfterCursorQuery
+				args = []driver.Value{"queue", 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", "first_seen_build_id"})
+				if !tt.empty {
+					var firstBuild any = issue.FirstSeenBuildID
+					if tt.scanError {
+						firstBuild = nil
+					}
+					rows.AddRow(issue.TaskID, issue.UnitID, firstBuild)
+					if tt.rowError {
+						rows.AddRow("task", "unit-2", "build/2").RowError(1, assert.AnError)
+					}
+				}
+				expectation.WillReturnRows(rows).RowsWillBeClosed()
+			}
+
+			got, err := store.List(context.Background(), storage.OpenIssueRange{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.OpenIssue{issue}, got)
+			}
+			require.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}
+
+func TestOpenIssueListRejectsInvalidRange(t *testing.T) {
+	for _, tt := range []struct {
+		name   string
+		queue  string
+		bounds storage.OpenIssueRange
+	}{
+		{name: "zero limit", bounds: storage.OpenIssueRange{}},
+		{name: "negative limit", bounds: storage.OpenIssueRange{Limit: -1}},
+		{name: "unit without task", bounds: storage.OpenIssueRange{Limit: 1,
+			After: storage.OpenIssueCursor{UnitID: "unit"}}},
+		{name: "oversized task", bounds: storage.OpenIssueRange{Limit: 1,
+			After: storage.OpenIssueCursor{TaskID: strings.Repeat("t", 256)}}},
+		{name: "oversized unit", bounds: storage.OpenIssueRange{Limit: 1,
+			After: storage.OpenIssueCursor{TaskID: "task", UnitID: strings.Repeat("😀", 257)}}},
+		{name: "oversized queue", queue: strings.Repeat("q", 256), bounds: storage.OpenIssueRange{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"
+			}
+			store := NewOpenIssueStore(db, testMetrics(), queue)
+			mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "first_seen_build_id"}))
+
+			got, err := store.List(context.Background(), tt.bounds)
+			require.Error(t, err)
+			require.Nil(t, got)
+		})
+	}
+}

stovepipe/extension/storage/mysql/open_issue_list_test.go

@@ -26,6 +26,17 @@ import (
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 )
 
+const listOpenIssuesQuery = `
+	SELECT task_id, unit_id, first_seen_build_id
+	FROM open_issue WHERE queue = ?
+	ORDER BY task_id ASC, unit_id ASC LIMIT ?`
+
+const listOpenIssuesAfterCursorQuery = `
+	SELECT task_id, unit_id, first_seen_build_id
+	FROM open_issue WHERE queue = ?
+		AND (task_id > ? OR (task_id = ? AND unit_id > ?))
+	ORDER BY task_id ASC, unit_id ASC LIMIT ?`
+
 type openIssueStore struct {
 	db    *sql.DB
 	scope tally.Scope
@@ -64,6 +75,9 @@ func (s *openIssueStore) Get(ctx context.Context, taskID, unitID string) (ret en
 	op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
 	defer func() { op.Complete(retErr) }()
 
+	if taskID == "" {
+		return entity.OpenIssue{}, fmt.Errorf("open issue task ID must not be empty")
+	}
 	if err := validateBuildIssueKey(s.queue, "", taskID, unitID); err != nil {
 		return entity.OpenIssue{}, err
 	}
@@ -84,6 +98,51 @@ func (s *openIssueStore) Get(ctx context.Context, taskID, unitID string) (ret en
 	return issue, nil
 }
 
+func (s *openIssueStore) List(ctx context.Context, bounds storage.OpenIssueRange) (ret []entity.OpenIssue, retErr error) {
+	op := metrics.Begin(s.scope, "list", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
+
+	cursor := bounds.After
+	if bounds.Limit <= 0 {
+		return nil, fmt.Errorf("open issue range requires a positive limit")
+	}
+	if cursor.TaskID == "" && cursor.UnitID != "" {
+		return nil, fmt.Errorf("open issue cursor requires a task ID when a unit ID is supplied")
+	}
+	if err := validateBuildIssueKey(s.queue, "", cursor.TaskID, cursor.UnitID); err != nil {
+		return nil, err
+	}
+
+	var (
+		rows *sql.Rows
+		err  error
+	)
+	if cursor.TaskID != "" {
+		rows, err = s.db.QueryContext(ctx, listOpenIssuesAfterCursorQuery,
+			s.queue, cursor.TaskID, cursor.TaskID, cursor.UnitID, bounds.Limit)
+	} else {
+		rows, err = s.db.QueryContext(ctx, listOpenIssuesQuery, s.queue, bounds.Limit)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("failed to list open issues: %w", err)
+	}
+	defer rows.Close()
+
+	issues := make([]entity.OpenIssue, 0)
+	for rows.Next() {
+		var issue entity.OpenIssue
+		if err := rows.Scan(&issue.TaskID, &issue.UnitID, &issue.FirstSeenBuildID); err != nil {
+			return nil, fmt.Errorf("failed to scan open issue: %w", err)
+		}
+		issues = append(issues, issue)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("failed to iterate open issues: %w", err)
+	}
+
+	return issues, nil
+}
+
 func (s *openIssueStore) Delete(ctx context.Context, issue entity.OpenIssue) (retErr error) {
 	op := metrics.Begin(s.scope, "delete", metrics.StorageLatencyBuckets)
 	defer func() { op.Complete(retErr) }()
@@ -104,6 +163,9 @@ func (s *openIssueStore) Delete(ctx context.Context, issue entity.OpenIssue) (re
 }
 
 func (s *openIssueStore) validateIssue(issue entity.OpenIssue) error {
+	if issue.TaskID == "" {
+		return fmt.Errorf("open issue task ID must not be empty")
+	}
 	if err := validateBuildIssueKey(s.queue, issue.FirstSeenBuildID, issue.TaskID, issue.UnitID); err != nil {
 		return err
 	}

stovepipe/extension/storage/mysql/open_issue_store.go

@@ -166,7 +166,7 @@ func TestOpenIssueDelete(t *testing.T) {
 }
 
 func TestOpenIssueRejectsInvalidIdentifiers(t *testing.T) {
-	for _, kind := range []string{"queue", "task", "unit", "first build", "empty first build"} {
+	for _, kind := range []string{"queue", "task", "empty task", "unit", "first build", "empty first build"} {
 		t.Run(kind, func(t *testing.T) {
 			db, mock, err := sqlmock.New()
 			require.NoError(t, err)
@@ -179,6 +179,8 @@ func TestOpenIssueRejectsInvalidIdentifiers(t *testing.T) {
 				queue = strings.Repeat("q", 256)
 			case "task":
 				issue.TaskID = strings.Repeat("t", 256)
+			case "empty task":
+				issue.TaskID = ""
 			case "unit":
 				issue.UnitID = strings.Repeat("😀", 257)
 			case "first build":
@@ -200,7 +202,7 @@ func TestOpenIssueRejectsInvalidIdentifiers(t *testing.T) {
 			require.Error(t, store.Delete(context.Background(), issue))
 
 			switch kind {
-			case "queue", "task", "unit":
+			case "queue", "task", "empty task", "unit":
 				_, err = store.Get(context.Background(), issue.TaskID, issue.UnitID)
 				require.Error(t, err)
 			}

stovepipe/extension/storage/mysql/open_issue_store_test.go

@@ -22,8 +22,25 @@ import (
 	"github.com/uber/submitqueue/stovepipe/entity"
 )
 
+// OpenIssueCursor is an exclusive position in ascending bytewise task/unit order.
+type OpenIssueCursor struct {
+	// TaskID is the task component; empty with an empty UnitID starts the range.
+	TaskID string
+	// UnitID is the unit component of the last returned key; empty is a task-wide issue.
+	UnitID string
+}
+
+// OpenIssueRange bounds a primary-key scan within the store's queue.
+type OpenIssueRange struct {
+	// After excludes this key and earlier keys; the zero cursor starts the range.
+	After OpenIssueCursor
+	// Limit is the positive maximum number of issues to return.
+	Limit int
+}
+
 // OpenIssueStore contains only currently open issues, keyed by task and unit
 // within the bound queue. A reopened issue must use a new first-seen build ID.
+// Issue keys require a nonempty task ID; an empty unit is a task-wide issue.
 type OpenIssueStore interface {
 	// Create requires a nonempty first-seen build ID.
 	// An existing task/unit returns ErrAlreadyExists.
@@ -32,6 +49,11 @@ type OpenIssueStore interface {
 	// Get returns ErrNotFound when the task/unit is not currently open.
 	Get(ctx context.Context, taskID, unitID string) (entity.OpenIssue, error)
 
+	// List returns at most Limit issues in ascending bytewise (task, unit) order.
+	// An empty slice ends the range; invalid bounds return an error.
+	// Pages are not a snapshot: concurrent creates/deletes may change membership.
+	List(ctx context.Context, bounds OpenIssueRange) ([]entity.OpenIssue, error)
+
 	// Delete removes only the matching opening. It succeeds if that opening is
 	// already absent and never removes a row with a different first-seen build.
 	Delete(ctx context.Context, issue entity.OpenIssue) error

stovepipe/extension/storage/open_issue_store.go

@@ -5,6 +5,7 @@ go_test(
     srcs = [
         "build_issue_test.go",
         "issue_log_test.go",
+        "open_issue_list_test.go",
         "open_issue_test.go",
         "project_fact_by_degree_test.go",
         "queue_state_transitions_test.go",