feat(stovepipe): associate build issues with repairs using entity CAS

stovepipe/entity/build_issue.go

@@ -14,11 +14,15 @@
 
 package entity
 
-// BuildIssue identifies a validation failure by task and optional unit.
+// BuildIssue is a validation failure observed within a build, identified by task and optional unit.
 type BuildIssue struct {
 	// TaskID identifies the logical validation task, stable across builds and retries; it must be nonempty.
 	TaskID string `json:"task_id"`
 	// UnitID identifies the failed unit within the task; empty for a task-wide issue.
 	// Repeated observations of the same condition use the same TaskID and UnitID.
 	UnitID string `json:"unit_id"`
+	// FixID is the opaque executor-provided repair identifier; empty means none is recorded.
+	FixID string `json:"fix_id"`
+	// Version is the optimistic-lock version, initially zero.
+	Version int32 `json:"version"`
 }

stovepipe/extension/storage/README.md

@@ -28,17 +28,21 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte
 
 ## 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. Build and task IDs must be nonempty; empty unit IDs represent task-wide observations.
+`BuildIssueStore` accumulates 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. Build and task IDs must be nonempty; empty unit IDs represent task-wide observations.
 
 Listing fixes the queue/build primary-key prefix and returns bounded pages in ascending bytewise task/unit order. A task/unit cursor continues after the last returned key; the zero cursor starts the range, and a unit without a task is invalid. Pages are not a snapshot across calls: concurrent observations can affect membership. Empty pages represent no observations in that range, not build success or completed analysis.
 
-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.
+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. Repair metadata is mutable; observation identities remain durable, with no deletion or absence-based recovery.
 
 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.
 
+`BuildIssue.FixID` associates an observation with an opaque executor-provided repair ID. An empty ID means no association is recorded. `Update` uses the existing entity CAS convention: it guards `oldVersion` and writes `newVersion`, returning `ErrVersionMismatch` for a stale or missing record. The caller owns version arithmetic and repair policy; storage does not enforce a per-build submission limit. Observation identity and membership survive recovery.
+
+Existing MySQL deployments must add `fix_id VARCHAR(255) NOT NULL DEFAULT ''` and `version INT NOT NULL DEFAULT 0` to `build_issue` before deploying the new readers and writers. The schema's `CREATE TABLE IF NOT EXISTS` does not alter existing tables.
+
 ## 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. 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.
+`OpenIssueStore` contains only currently open issues, keyed by task and unit within the bound queue. Removing an open row never removes durable 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.
 

stovepipe/extension/storage/build_issue_store.go

@@ -38,7 +38,7 @@ type BuildIssueRange struct {
 	Limit int
 }
 
-// BuildIssueStore retains immutable observations keyed by build, task, and unit
+// BuildIssueStore retains observations and their repair association keyed by build, task, and unit
 // within the bound queue. Presence records an observation, not a currently open issue.
 // Observation keys require nonempty build and task IDs; an empty unit is task-wide.
 type BuildIssueStore interface {
@@ -53,4 +53,9 @@ type BuildIssueStore interface {
 	// An empty slice ends the range; invalid bounds or an empty build ID return an error.
 	// Pages are not a snapshot: concurrent observations may change membership.
 	List(ctx context.Context, buildID string, bounds BuildIssueRange) ([]entity.BuildIssue, error)
+
+	// Update persists the mutable fields if the stored version matches oldVersion,
+	// writing newVersion. Returns ErrVersionMismatch on a stale or missing record.
+	// The caller owns version arithmetic; the store does not read issue.Version.
+	Update(ctx context.Context, buildID string, issue entity.BuildIssue, oldVersion, newVersion int32) error
 }

stovepipe/extension/storage/mysql/BUILD.bazel

@@ -35,6 +35,7 @@ go_test(
     srcs = [
         "build_issue_list_test.go",
         "build_issue_store_test.go",
+        "build_issue_update_test.go",
         "build_store_test.go",
         "issue_log_store_test.go",
         "open_issue_list_test.go",

stovepipe/extension/storage/mysql/build_issue_list_test.go

@@ -50,7 +50,7 @@ func TestBuildIssueList(t *testing.T) {
 			require.NoError(t, err)
 			defer db.Close()
 			store := NewBuildIssueStore(db, testMetrics(), "queue")
-			issue := entity.BuildIssue{TaskID: "task", UnitID: "unit"}
+			issue := entity.BuildIssue{TaskID: "task", UnitID: "unit", FixID: "Fix/1 ", Version: 3}
 			query := listBuildIssuesQuery
 			args := []driver.Value{"queue", "build/1", int64(2)}
 			if tt.after.TaskID != "" {
@@ -61,15 +61,15 @@ func TestBuildIssueList(t *testing.T) {
 			if tt.dbErr != nil {
 				expectation.WillReturnError(tt.dbErr)
 			} else {
-				rows := sqlmock.NewRows([]string{"task_id", "unit_id"})
+				rows := sqlmock.NewRows([]string{"task_id", "unit_id", "fix_id", "version"})
 				if !tt.empty {
 					var task any = issue.TaskID
 					if tt.scanError {
 						task = nil
 					}
-					rows.AddRow(task, issue.UnitID)
+					rows.AddRow(task, issue.UnitID, issue.FixID, issue.Version)
 					if tt.rowError {
-						rows.AddRow("task", "unit-2").RowError(1, assert.AnError)
+						rows.AddRow("task", "unit-2", "", 0).RowError(1, assert.AnError)
 					}
 				}
 				expectation.WillReturnRows(rows).RowsWillBeClosed()
@@ -128,7 +128,7 @@ func TestBuildIssueListRejectsInvalidRange(t *testing.T) {
 			}
 			buildID := tt.buildID
 			store := NewBuildIssueStore(db, testMetrics(), queue)
-			mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id"}))
+			mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "fix_id", "version"}))
 
 			got, err := store.List(context.Background(), buildID, tt.bounds)
 			require.Error(t, err)

stovepipe/extension/storage/mysql/build_issue_store.go

@@ -28,12 +28,12 @@ import (
 )
 
 const listBuildIssuesQuery = `
-	SELECT task_id, unit_id
+	SELECT task_id, unit_id, fix_id, version
 	FROM build_issue WHERE queue = ? AND build_id = ?
 	ORDER BY task_id ASC, unit_id ASC LIMIT ?`
 
 const listBuildIssuesAfterCursorQuery = `
-	SELECT task_id, unit_id
+	SELECT task_id, unit_id, fix_id, version
 	FROM build_issue WHERE queue = ? AND build_id = ?
 		AND (task_id > ? OR (task_id = ? AND unit_id > ?))
 	ORDER BY task_id ASC, unit_id ASC LIMIT ?`
@@ -57,9 +57,9 @@ func (s *buildIssueStore) Create(ctx context.Context, buildID string, issue enti
 		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,
+		`INSERT INTO build_issue (queue, build_id, task_id, unit_id, fix_id, version)
+         VALUES (?, ?, ?, ?, ?, ?)`,
+		s.queue, buildID, issue.TaskID, issue.UnitID, issue.FixID, issue.Version,
 	)
 	if err != nil {
 		if isDuplicateEntry(err) {
@@ -82,10 +82,10 @@ func (s *buildIssueStore) Get(ctx context.Context, buildID, taskID, unitID strin
 
 	var issue entity.BuildIssue
 	err := s.db.QueryRowContext(ctx,
-		`SELECT task_id, unit_id FROM build_issue
+		`SELECT task_id, unit_id, fix_id, version FROM build_issue
          WHERE queue = ? AND build_id = ? AND task_id = ? AND unit_id = ?`,
 		s.queue, buildID, taskID, unitID,
-	).Scan(&issue.TaskID, &issue.UnitID)
+	).Scan(&issue.TaskID, &issue.UnitID, &issue.FixID, &issue.Version)
 	if errors.Is(err, sql.ErrNoRows) {
 		return entity.BuildIssue{}, storage.WrapNotFound(err)
 	}
@@ -130,7 +130,7 @@ func (s *buildIssueStore) List(ctx context.Context, buildID string, bounds stora
 	issues := make([]entity.BuildIssue, 0)
 	for rows.Next() {
 		var issue entity.BuildIssue
-		if err := rows.Scan(&issue.TaskID, &issue.UnitID); err != nil {
+		if err := rows.Scan(&issue.TaskID, &issue.UnitID, &issue.FixID, &issue.Version); err != nil {
 			return nil, fmt.Errorf("failed to scan build issue: %w", err)
 		}
 		issues = append(issues, issue)
@@ -142,6 +142,31 @@ func (s *buildIssueStore) List(ctx context.Context, buildID string, bounds stora
 	return issues, nil
 }
 
+func (s *buildIssueStore) Update(ctx context.Context, buildID string, issue entity.BuildIssue, oldVersion, newVersion int32) (retErr error) {
+	op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
+
+	if err := s.validateKey(buildID, issue.TaskID, issue.UnitID); err != nil {
+		return err
+	}
+	result, err := s.db.ExecContext(ctx,
+		`UPDATE build_issue SET fix_id = ?, version = ?
+         WHERE queue = ? AND build_id = ? AND task_id = ? AND unit_id = ? AND version = ?`,
+		issue.FixID, newVersion, s.queue, buildID, issue.TaskID, issue.UnitID, oldVersion)
+	if err != nil {
+		return fmt.Errorf("failed to update build issue: %w", err)
+	}
+	affected, err := result.RowsAffected()
+	if err != nil {
+		return fmt.Errorf("failed to get build issue rows affected: %w", err)
+	}
+	if affected != 1 {
+		return fmt.Errorf("build issue build=%q task=%q unit=%q expected_version=%d: %w",
+			buildID, issue.TaskID, issue.UnitID, oldVersion, storage.ErrVersionMismatch)
+	}
+	return nil
+}
+
 func (s *buildIssueStore) validateKey(buildID, taskID, unitID string) error {
 	if buildID == "" || taskID == "" {
 		return fmt.Errorf("build issue requires nonempty build and task IDs")

stovepipe/extension/storage/mysql/build_issue_store_test.go

@@ -53,8 +53,8 @@ func TestBuildIssueStoreCreate(t *testing.T) {
 			store := NewBuildIssueStore(db, testMetrics(), "queue")
 			issue := entity.BuildIssue{TaskID: "task", UnitID: tt.unitID}
 
-			expectation := mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO build_issue (queue, build_id, task_id, unit_id) VALUES (?, ?, ?, ?)`)).
-				WithArgs("queue", "build", issue.TaskID, issue.UnitID)
+			expectation := mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO build_issue (queue, build_id, task_id, unit_id, fix_id, version) VALUES (?, ?, ?, ?, ?, ?)`)).
+				WithArgs("queue", "build", issue.TaskID, issue.UnitID, issue.FixID, issue.Version)
 			if tt.dbErr != nil {
 				expectation.WillReturnError(tt.dbErr)
 			} else {
@@ -97,14 +97,14 @@ func TestBuildIssueStoreGet(t *testing.T) {
 			store := NewBuildIssueStore(db, testMetrics(), "queue")
 			issue := entity.BuildIssue{TaskID: "task", UnitID: tt.unitID}
 
-			expectation := mock.ExpectQuery(regexp.QuoteMeta(`SELECT task_id, unit_id FROM build_issue WHERE queue = ? AND build_id = ? AND task_id = ? AND unit_id = ?`)).
+			expectation := mock.ExpectQuery(regexp.QuoteMeta(`SELECT task_id, unit_id, fix_id, version FROM build_issue WHERE queue = ? AND build_id = ? AND task_id = ? AND unit_id = ?`)).
 				WithArgs("queue", "build", issue.TaskID, issue.UnitID)
 			if tt.queryErr != nil {
 				expectation.WillReturnError(tt.queryErr)
 			} else if tt.scanErr {
-				expectation.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id"}).AddRow(nil, issue.UnitID))
+				expectation.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "fix_id", "version"}).AddRow(nil, issue.UnitID, "", 0))
 			} else {
-				expectation.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id"}).AddRow(issue.TaskID, issue.UnitID))
+				expectation.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "fix_id", "version"}).AddRow(issue.TaskID, issue.UnitID, issue.FixID, issue.Version))
 			}
 
 			got, err := store.Get(context.Background(), "build", issue.TaskID, issue.UnitID)
@@ -150,15 +150,16 @@ func TestBuildIssueStoreRejectsOversizedIdentifiers(t *testing.T) {
 			// SQL would succeed, so rejection must come from identifier validation.
 			mock.MatchExpectationsInOrder(false)
 			mock.ExpectExec("INSERT INTO build_issue").WillReturnResult(sqlmock.NewResult(0, 1))
-			mock.ExpectQuery("SELECT task_id, unit_id FROM build_issue").WillReturnRows(
-				sqlmock.NewRows([]string{"task_id", "unit_id"}).AddRow(tt.issue.TaskID, tt.issue.UnitID),
+			mock.ExpectQuery("SELECT").WillReturnRows(
+				sqlmock.NewRows([]string{"task_id", "unit_id", "fix_id", "version"}).AddRow(tt.issue.TaskID, tt.issue.UnitID, tt.issue.FixID, tt.issue.Version),
 			)
 
 			require.Error(t, store.Create(context.Background(), tt.buildID, tt.issue))
 
 			got, err := store.Get(context.Background(), tt.buildID, tt.issue.TaskID, tt.issue.UnitID)
 			require.Error(t, err)
 			require.Equal(t, entity.BuildIssue{}, got)
+			require.Error(t, store.Update(context.Background(), tt.buildID, tt.issue, 0, 1))
 		})
 	}
 }
@@ -179,7 +180,7 @@ func TestBuildIssueStoreRejectsMissingIdentifiers(t *testing.T) {
 			store := NewBuildIssueStore(db, testMetrics(), "queue")
 			mock.MatchExpectationsInOrder(false)
 			mock.ExpectExec("INSERT INTO build_issue").WillReturnResult(sqlmock.NewResult(0, 1))
-			mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id"}).AddRow(tt.taskID, "unit"))
+			mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "fix_id", "version"}).AddRow(tt.taskID, "unit", "", 0))
 
 			require.Error(t, store.Create(context.Background(), tt.buildID, entity.BuildIssue{TaskID: tt.taskID, UnitID: "unit"}))
 			_, err = store.Get(context.Background(), tt.buildID, tt.taskID, "unit")

stovepipe/extension/storage/mysql/build_issue_update_test.go

@@ -0,0 +1,81 @@
+// 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"
+	"regexp"
+	"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 TestBuildIssueUpdate(t *testing.T) {
+	for _, tt := range []struct {
+		name     string
+		affected int64
+		execErr  error
+		rowsErr  error
+		wantErr  error
+	}{
+		{name: "updated", affected: 1},
+		{name: "stale or missing", wantErr: storage.ErrVersionMismatch},
+		{name: "write failure", execErr: assert.AnError, wantErr: assert.AnError},
+		{name: "rows affected failure", rowsErr: assert.AnError, wantErr: assert.AnError},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			db, mock, err := sqlmock.New()
+			require.NoError(t, err)
+			defer db.Close()
+			issue := entity.BuildIssue{TaskID: "Task ", UnitID: "Unit ", FixID: "Fix/1 ", Version: 99}
+			original := issue
+			expectation := mock.ExpectExec(regexp.QuoteMeta(`UPDATE build_issue SET fix_id = ?, version = ?
+                WHERE queue = ? AND build_id = ? AND task_id = ? AND unit_id = ? AND version = ?`)).
+				WithArgs(issue.FixID, int32(2), "Queue ", "Build ", issue.TaskID, issue.UnitID, int32(1))
+			switch {
+			case tt.execErr != nil:
+				expectation.WillReturnError(tt.execErr)
+			case tt.rowsErr != nil:
+				expectation.WillReturnResult(sqlmock.NewErrorResult(tt.rowsErr))
+			default:
+				expectation.WillReturnResult(sqlmock.NewResult(0, tt.affected))
+			}
+			err = NewBuildIssueStore(db, testMetrics(), "Queue ").Update(context.Background(), "Build ", issue, 1, 2)
+			if tt.wantErr != nil {
+				require.ErrorIs(t, err, tt.wantErr)
+			} else {
+				require.NoError(t, err)
+			}
+			require.Equal(t, original, issue)
+			require.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}
+
+func TestBuildIssueUpdateRejectsMissingIdentity(t *testing.T) {
+	for _, key := range []struct{ buildID, taskID string }{{taskID: "task"}, {buildID: "build"}} {
+		db, mock, err := sqlmock.New()
+		require.NoError(t, err)
+		store := NewBuildIssueStore(db, testMetrics(), "queue")
+		require.Error(t, store.Update(context.Background(), key.buildID, entity.BuildIssue{TaskID: key.taskID}, 0, 1))
+		require.NoError(t, mock.ExpectationsWereMet())
+		mock.ExpectClose()
+		require.NoError(t, db.Close())
+	}
+}

stovepipe/extension/storage/mysql/schema/build_issue.sql

@@ -1,4 +1,4 @@
--- One immutable observation per (queue, build, task, unit), not an open-issue projection.
+-- Durable observation identity with a versioned repair association; not an open-issue projection.
 -- Binary identifiers preserve byte-exact identity; the full key is at most 2554 bytes.
 CREATE TABLE IF NOT EXISTS build_issue (
     queue       VARBINARY(255) NOT NULL,
@@ -7,5 +7,8 @@ CREATE TABLE IF NOT EXISTS build_issue (
     task_id     VARBINARY(255) NOT NULL,
     unit_id     VARBINARY(1024) NOT NULL,
 
+    fix_id      VARCHAR(255) NOT NULL DEFAULT '',
+    version     INT NOT NULL DEFAULT 0,
+
     PRIMARY KEY (queue, build_id, task_id, unit_id)
 ) ENGINE=InnoDB;

test/integration/stovepipe/extension/storage/mysql/BUILD.bazel

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