feat(stovepipe): Add open issue lookup storage

stovepipe/extension/storage/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "issue_state_store.go",
         "issue_summary_store.go",
         "issue_tracking_store.go",
+        "open_issue_store.go",
         "project_fact_by_degree_store.go",
         "queue_log_store.go",
         "queue_store.go",

stovepipe/extension/storage/README.md

@@ -48,7 +48,7 @@ Create is conditional on absence, including closed records. Update checks the su
 
 Listing reads a bounded primary-key range in bytewise task/unit order and includes both statuses. Callers filter closed records in memory and advance after the last scanned key, including closed records. A page containing only closed records is not the end of the range. No status index or server-side filtering is required. Task IDs must be nonempty; empty units identify task-wide issues. Identifier byte limits match the build-issue store.
 
-The MySQL backend uses the new `issue_state` table. Create it before deploying the updated readers. The prior `open_issue` table is not read or automatically copied; no in-repository runtime writer was enabled for it. Deployments with externally populated rows need an explicit backfill before switching readers. The repair executor still receives an `OpenIssue`, which describes an open occurrence rather than the storage record.
+The MySQL backend uses `issue_state` for operational tracking. The separate `open_issue` table is a derived lookup, not a replacement for retained issue state. The repair executor still receives an `OpenIssue`, which describes an open occurrence rather than the operational storage record.
 
 ## Issue summaries
 
@@ -58,6 +58,16 @@ The MySQL backend uses the new `issue_state` table. Create it before deploying t
 
 The MySQL primary key is `(queue, task_id, unit_id)` with byte-exact identity and the same identifier limits as the issue-state store. Source request IDs follow the existing canonical positive decimal ID contract. The new `issue_summary` table must exist before deploying readers or writers. The [issue-log materializer](../../core/issuelog/README.md) updates summaries during normal issue tracking and recovery. No public open-issues list is exposed yet.
 
+## Open-issue lookups
+
+`OpenIssueStore` exposes the existing `open_issue` table as a queue-scoped lookup keyed by task/unit. Each open-issue row contains only the opening's `FirstSeenBuildID`, using the existing `OpenIssue` entity and identifier byte limits. No columns, indexes, or operational issue-state fields are added.
+
+Create returns `ErrAlreadyExists` for any existing task/unit key. Replace compares the expected opening build ID before changing it; Delete compares the opening in the supplied open-issue row before removing it. Both operations return whether they changed the open-issue row, with false for a missing key or a different opening. Replacing an opening with itself is also a no-op returning false. These are atomic conditional writes to one complete key, not build-order checks. An old close that still expects a previous opening cannot delete a newer opening.
+
+Listing traverses a bounded queue-prefix primary-key range in ascending bytewise task/unit order. The zero cursor starts the range; otherwise the cursor excludes its key and all earlier keys, even if that open-issue row has been deleted. Empty units identify task-wide issues, while task IDs must be nonempty. Membership can change between reads. A consumer that skips stale references must still advance after the last returned open-issue row, not the last accepted issue.
+
+Open-issue rows are not authoritative issue status. A delayed create can leave a reference to a closed issue, so readers must verify that the summary is open and its opening matches before returning a result. The store does not read summaries, enforce source ordering, backfill existing rows, or populate open-issue rows automatically. No runtime writer or public list API uses this store yet.
+
 ## Issue lifecycle history
 
 `IssueLogStore` retains immutable open/closed status transitions independently of current issue state and per-build observations. Each transition is keyed by task, unit, build, and status within the bound queue. An empty unit identifies a task-wide issue. History remains when an issue closes or reopens; transitions do not carry a separate reference to the opening they belong to.

stovepipe/extension/storage/mock/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "issue_state_store_mock.go",
         "issue_summary_store_mock.go",
         "issue_tracking_store_mock.go",
+        "open_issue_store_mock.go",
         "project_fact_by_degree_store_mock.go",
         "queue_log_store_mock.go",
         "queue_store_mock.go",

stovepipe/extension/storage/mysql/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "issue_state_store.go",
         "issue_summary_store.go",
         "issue_tracking_store.go",
+        "open_issue_store.go",
         "project_fact_by_degree_store.go",
         "queue_log_store.go",
         "queue_store.go",
@@ -44,6 +45,7 @@ go_test(
         "issue_state_store_test.go",
         "issue_summary_store_test.go",
         "issue_tracking_store_test.go",
+        "open_issue_store_test.go",
         "project_fact_by_degree_store_test.go",
         "queue_store_test.go",
         "queue_summary_store_test.go",

stovepipe/extension/storage/mysql/open_issue_store.go

@@ -0,0 +1,194 @@
+// 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"
+	"errors"
+	"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"
+)
+
+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
+	queue string
+}
+
+// NewOpenIssueStore binds derived open-issue open issues to one queue.
+func NewOpenIssueStore(db *sql.DB, scope tally.Scope, queue string) storage.OpenIssueStore {
+	return &openIssueStore{db: db, scope: scope, queue: queue}
+}
+
+func (s *openIssueStore) Create(ctx context.Context, issue entity.OpenIssue) (retErr error) {
+	op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
+
+	if err := s.validateIssue(issue); err != nil {
+		return err
+	}
+	_, err := s.db.ExecContext(ctx,
+		`INSERT INTO open_issue (queue, task_id, unit_id, first_seen_build_id) VALUES (?, ?, ?, ?)`,
+		s.queue, issue.TaskID, issue.UnitID, issue.FirstSeenBuildID)
+	if isDuplicateEntry(err) {
+		return storage.ErrAlreadyExists
+	}
+	if err != nil {
+		return fmt.Errorf("create open issue open issue: %w", err)
+	}
+	return nil
+}
+
+func (s *openIssueStore) Get(ctx context.Context, taskID, unitID string) (issue entity.OpenIssue, retErr error) {
+	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
+	}
+	err := s.db.QueryRowContext(ctx,
+		`SELECT task_id, unit_id, first_seen_build_id FROM open_issue WHERE queue = ? AND task_id = ? AND unit_id = ?`,
+		s.queue, taskID, unitID,
+	).Scan(&issue.TaskID, &issue.UnitID, &issue.FirstSeenBuildID)
+	if errors.Is(err, sql.ErrNoRows) {
+		return entity.OpenIssue{}, storage.WrapNotFound(err)
+	}
+	if err != nil {
+		return entity.OpenIssue{}, fmt.Errorf("get open issue open issue: %w", err)
+	}
+	return issue, nil
+}
+
+func (s *openIssueStore) Replace(ctx context.Context, issue entity.OpenIssue, expectedFirstSeenBuildID string) (replaced bool, retErr error) {
+	op := metrics.Begin(s.scope, "replace", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
+
+	if err := s.validateIssue(issue); err != nil {
+		return false, err
+	}
+	if expectedFirstSeenBuildID == "" {
+		return false, fmt.Errorf("open issue replacement requires an expected opening build ID")
+	}
+	if err := validateBuildIssueKey(s.queue, expectedFirstSeenBuildID, issue.TaskID, issue.UnitID); err != nil {
+		return false, err
+	}
+	if issue.FirstSeenBuildID == expectedFirstSeenBuildID {
+		return false, nil
+	}
+
+	result, err := s.db.ExecContext(ctx,
+		`UPDATE open_issue SET first_seen_build_id = ? WHERE queue = ? AND task_id = ? AND unit_id = ? AND first_seen_build_id = ?`,
+		issue.FirstSeenBuildID, s.queue, issue.TaskID, issue.UnitID, expectedFirstSeenBuildID)
+	if err != nil {
+		return false, fmt.Errorf("replace open issue open issue: %w", err)
+	}
+	rows, err := result.RowsAffected()
+	if err != nil {
+		return false, fmt.Errorf("read open issue replacement count: %w", err)
+	}
+	return rows == 1, nil
+}
+
+func (s *openIssueStore) Delete(ctx context.Context, issue entity.OpenIssue) (deleted bool, retErr error) {
+	op := metrics.Begin(s.scope, "delete", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
+
+	if err := s.validateIssue(issue); err != nil {
+		return false, err
+	}
+	result, err := s.db.ExecContext(ctx,
+		`DELETE FROM open_issue WHERE queue = ? AND task_id = ? AND unit_id = ? AND first_seen_build_id = ?`,
+		s.queue, issue.TaskID, issue.UnitID, issue.FirstSeenBuildID)
+	if err != nil {
+		return false, fmt.Errorf("delete open issue open issue: %w", err)
+	}
+	rows, err := result.RowsAffected()
+	if err != nil {
+		return false, fmt.Errorf("read open issue deletion count: %w", err)
+	}
+	return rows == 1, 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("list open issue 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("scan open issue open issue: %w", err)
+		}
+		issues = append(issues, issue)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("iterate open issue open issues: %w", err)
+	}
+	return issues, nil
+}
+
+func (s *openIssueStore) validateIssue(issue entity.OpenIssue) error {
+	if issue.TaskID == "" || issue.FirstSeenBuildID == "" {
+		return fmt.Errorf("open issue open issue requires task and opening build IDs")
+	}
+	return validateBuildIssueKey(s.queue, issue.FirstSeenBuildID, issue.TaskID, issue.UnitID)
+}
+
+var _ storage.OpenIssueStore = (*openIssueStore)(nil)