feat(stovepipe): Store currently open issues

stovepipe/entity/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "build_issue.go",
         "ingest.go",
         "list.go",
+        "open_issue.go",
         "project_status.go",
         "queue.go",
         "queue_config.go",

stovepipe/entity/open_issue.go

@@ -0,0 +1,27 @@
+// 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
+
+// OpenIssue is a currently open validation issue within a queue.
+// 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 string `json:"task_id"`
+	// UnitID identifies the validation unit; empty for a task-wide issue.
+	UnitID string `json:"unit_id"`
+	// FirstSeenBuildID identifies the build where this opening of the issue was first observed.
+	FirstSeenBuildID string `json:"first_seen_build_id"`
+}

stovepipe/extension/storage/BUILD.bazel

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "build_issue_store.go",
         "build_store.go",
+        "open_issue_store.go",
         "queue_log_store.go",
         "queue_store.go",
         "queue_summary_store.go",

stovepipe/extension/storage/README.md

@@ -1,6 +1,6 @@
 # Storage
 
-Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`, `BuildIssueStore`). Implementations live under `extension/storage/<impl>/`.
+Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`, `BuildIssueStore`, `OpenIssueStore`). Implementations live under `extension/storage/<impl>/`.
 
 The aggregate is resolved per queue through a factory keyed by queue name, mirroring the extension contract and the [submitqueue storage seam](../../../submitqueue/extension/storage/README.md#queue-scoped-resolution): a resolved instance is bound to its queue, entity arguments whose queue disagrees with the binding are rejected, and reads never surface another queue's records. Stovepipe has no cross-queue read paths, so every store — including `QueueStore`, whose row key is the queue name itself — lives inside the queue-scoped aggregate; there is no global remainder.
 
@@ -31,3 +31,13 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte
 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.
 
 The MySQL primary key contains the queue, build ID, task ID, and unit ID directly as binary strings, preserving byte-exact identity. Their limits are 255, 1020, 255, and 1024 bytes respectively, for a combined maximum of 2554 bytes. Creates and reads reject identifiers exceeding these limits before querying the database; identifiers are never truncated.
+
+## 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.
+
+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.
+
+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. History and source-order reconciliation belong to later controllers, not this membership store.

stovepipe/extension/storage/mock/BUILD.bazel

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "build_issue_store_mock.go",
         "build_store_mock.go",
+        "open_issue_store_mock.go",
         "queue_log_store_mock.go",
         "queue_store_mock.go",
         "queue_summary_store_mock.go",

stovepipe/extension/storage/mysql/BUILD.bazel

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "build_issue_store.go",
         "build_store.go",
+        "open_issue_store.go",
         "queue_log_store.go",
         "queue_store.go",
         "queue_summary_store.go",
@@ -32,6 +33,7 @@ go_test(
     srcs = [
         "build_issue_store_test.go",
         "build_store_test.go",
+        "open_issue_store_test.go",
         "queue_store_test.go",
         "queue_summary_store_test.go",
         "request_acceptance_store_test.go",

stovepipe/extension/storage/mysql/open_issue_store.go

@@ -0,0 +1,114 @@
+// 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"
+)
+
+type openIssueStore struct {
+	db    *sql.DB
+	scope tally.Scope
+	queue string
+}
+
+// NewOpenIssueStore binds currently open issues to queue over the shared database.
+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 err != nil {
+		if isDuplicateEntry(err) {
+			return fmt.Errorf("open issue task=%q unit=%q: %w", issue.TaskID, issue.UnitID, storage.ErrAlreadyExists)
+		}
+
+		return fmt.Errorf("failed to create open issue: %w", err)
+	}
+
+	return nil
+}
+
+func (s *openIssueStore) Get(ctx context.Context, taskID, unitID string) (ret entity.OpenIssue, retErr error) {
+	op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
+
+	if err := validateBuildIssueKey(s.queue, "", taskID, unitID); err != nil {
+		return entity.OpenIssue{}, err
+	}
+
+	var issue entity.OpenIssue
+	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("failed to get open issue: %w", err)
+	}
+
+	return issue, 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) }()
+
+	if err := s.validateIssue(issue); err != nil {
+		return err
+	}
+
+	_, 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 fmt.Errorf("failed to delete open issue: %w", err)
+	}
+
+	return nil
+}
+
+func (s *openIssueStore) validateIssue(issue entity.OpenIssue) error {
+	if err := validateBuildIssueKey(s.queue, issue.FirstSeenBuildID, issue.TaskID, issue.UnitID); err != nil {
+		return err
+	}
+	if issue.FirstSeenBuildID == "" {
+		return fmt.Errorf("open issue first-seen build ID must not be empty")
+	}
+	return nil
+}