feat(stovepipe): Persist build issue observations (#902)

## Summary

**What**:

- Persist individual validation issue observations per build and queue,
with duplicate detection.

**Why**:

- Retain observed failures across polls and restarts without replacing
earlier observations or rewriting serialized collections.

## Test Plan

- [x] Add unit tests.
- [x] Verify persistence against a real MySQL database.

## Revert Plan

- Revert this PR. No runtime consumer is enabled; the additive table can
remain unused without affecting validation.

## Issue Links

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

## Stack

1. #902
1. #907
1. #915
1. #909
1. #916

Author: prathshenoy

stovepipe/extension/storage/BUILD.bazel

@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = [
+        "build_issue_store.go",
         "build_store.go",
         "queue_log_store.go",
         "queue_store.go",

stovepipe/extension/storage/README.md

@@ -1,6 +1,6 @@
 # Storage
 
-Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`). Implementations live under `extension/storage/<impl>/`.
+Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`, `BuildIssueStore`). 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.
 
@@ -23,3 +23,11 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte
 `RequestAcceptanceStore` is the ordered lookup mapping for List: immutable `(queue, accepted_at_ms, request_id)` keys with bounded primary-key range scans, not a secondary index or a second status projection. The request-log materializer ensures each known acceptance has a mapping after summary persistence, including retries where the summary is unchanged. Unknown acceptance times are not mapped; historical backfill is separate.
 
 `QueueStore` owns internal enablement alongside pipeline coordination. `QueueSummaryStore` persists the current read model projected from immutable `QueueLogStore` occurrences. Enablement sequences, operational row versions and projection versions are distinct. These stores use primary-key operations; version arithmetic, duplicate reconciliation and projection repair belong to business logic. See [queue state](../../../doc/rfc/stovepipe/queue-state.md).
+
+## 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.
+
+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.

stovepipe/extension/storage/build_issue_store.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 storage
+
+//go:generate mockgen -source=build_issue_store.go -destination=mock/build_issue_store_mock.go -package=mock
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/stovepipe/entity"
+)
+
+// BuildIssueStore retains immutable observations keyed by build, task, and unit
+// within the bound queue. Presence records an observation, not a currently open issue.
+type BuildIssueStore interface {
+	// Create returns ErrAlreadyExists for a previously recorded build/task/unit.
+	// Callers may treat that duplicate as success when accumulating observations.
+	Create(ctx context.Context, buildID string, issue entity.BuildIssue) error
+
+	// Get returns ErrNotFound when the build/task/unit has not been recorded.
+	Get(ctx context.Context, buildID, taskID, unitID string) (entity.BuildIssue, error)
+}

stovepipe/extension/storage/mock/BUILD.bazel

@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library")
 go_library(
     name = "go_default_library",
     srcs = [
+        "build_issue_store_mock.go",
         "build_store_mock.go",
         "queue_log_store_mock.go",
         "queue_store_mock.go",

stovepipe/extension/storage/mock/build_issue_store_mock.go

@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = [
+        "build_issue_store.go",
         "build_store.go",
         "queue_log_store.go",
         "queue_store.go",
@@ -29,6 +30,7 @@ go_library(
 go_test(
     name = "go_default_test",
     srcs = [
+        "build_issue_store_test.go",
         "build_store_test.go",
         "queue_store_test.go",
         "queue_summary_store_test.go",

stovepipe/extension/storage/mock/storage_mock.go

@@ -0,0 +1,106 @@
+// 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 buildIssueStore struct {
+	db    *sql.DB
+	scope tally.Scope
+	queue string
+}
+
+// NewBuildIssueStore binds observations to queue over the shared database.
+func NewBuildIssueStore(db *sql.DB, scope tally.Scope, queue string) storage.BuildIssueStore {
+	return &buildIssueStore{db: db, scope: scope, queue: queue}
+}
+
+func (s *buildIssueStore) Create(ctx context.Context, buildID string, issue entity.BuildIssue) (retErr error) {
+	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 {
+		return err
+	}
+	_, err := s.db.ExecContext(ctx,
+		`INSERT INTO build_issue (queue, build_id, task_id, unit_id)
+         VALUES (?, ?, ?, ?)`,
+		s.queue, buildID, issue.TaskID, issue.UnitID,
+	)
+	if err != nil {
+		if isDuplicateEntry(err) {
+			return fmt.Errorf("build issue build=%q task=%q unit=%q: %w", buildID, issue.TaskID, issue.UnitID, storage.ErrAlreadyExists)
+		}
+
+		return fmt.Errorf("failed to create build issue build=%q task=%q unit=%q: %w", buildID, issue.TaskID, issue.UnitID, err)
+	}
+
+	return nil
+}
+
+func (s *buildIssueStore) Get(ctx context.Context, buildID, taskID, unitID string) (ret entity.BuildIssue, retErr error) {
+	op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
+
+	if err := validateBuildIssueKey(s.queue, buildID, taskID, unitID); err != nil {
+		return entity.BuildIssue{}, err
+	}
+
+	var issue entity.BuildIssue
+	err := s.db.QueryRowContext(ctx,
+		`SELECT task_id, unit_id FROM build_issue
+         WHERE queue = ? AND build_id = ? AND task_id = ? AND unit_id = ?`,
+		s.queue, buildID, taskID, unitID,
+	).Scan(&issue.TaskID, &issue.UnitID)
+	if errors.Is(err, sql.ErrNoRows) {
+		return entity.BuildIssue{}, storage.WrapNotFound(err)
+	}
+
+	if err != nil {
+		return entity.BuildIssue{}, fmt.Errorf("failed to get build issue build=%q task=%q unit=%q: %w", buildID, taskID, unitID, err)
+	}
+
+	return issue, nil
+}
+
+func validateBuildIssueKey(queue, buildID, taskID, unitID string) error {
+	for _, field := range []struct {
+		name     string
+		value    string
+		maxBytes int
+	}{
+		{"queue", queue, 255},
+		{"build ID", buildID, 1020},
+		{"task ID", taskID, 255},
+		{"unit ID", unitID, 1024},
+	} {
+		if len(field.value) > field.maxBytes {
+			return fmt.Errorf("build issue %s exceeds %d-byte limit", field.name, field.maxBytes)
+		}
+	}
+
+	return nil
+}