feat(stovepipe): persist repairs with queue-scoped discovery

Author: mnoah1

doc/rfc/stovepipe/fixer.md

@@ -85,4 +85,4 @@ The fixer adds one table and uses primary-key operations. Build-analysis history
 
 Both queue controllers follow [`consumer.Controller`](../../../platform/consumer/controller.go): `Process(context.Context, consumer.Delivery) error`, `Name`, `TopicKey`, and `ConsumerGroup`. Service wiring resolves per-queue evaluators/executors, registers fix/fixsignal and their recovery behavior, and supplies host integrations.
 
-The API layer defines the contracts; the persistence layer supplies MySQL support. Runtime controller wiring and provider adapters follow as separate changes. Initial dispatch can run before complete recovery reconciliation, while leaving completion awaiting verification.
+The API layer defines the contracts; this persistence layer supplies MySQL support. Runtime controller wiring and provider adapters follow as separate changes. Initial dispatch can run before complete recovery reconciliation, while leaving completion awaiting verification.

stovepipe/extension/storage/README.md

@@ -51,4 +51,4 @@ Resolution belongs to a later controller: a successful, authoritative final anal
 
 `OpenBuildIssue.FixID` is the authoritative assignment for its opening. Updates replace last-observed build and assignment together under the opening/version guards. Analysis preserves assignments; fixer preserves observation fields. Resolution history and recovery evidence remain owned by build analysis; deleting a current row alone supplies no validating build reference.
 
-The [repair RFC](../../../doc/rfc/stovepipe/fixer.md) defines persistence, publication, and observation ordering. Persistence implementations follow in the storage layer.
+The [repair RFC](../../../doc/rfc/stovepipe/fixer.md) defines persistence, publication, and observation ordering. MySQL persists these fields through complete, version-guarded snapshots. Before deployment, add/backfill `build.first_fix_id`, `open_build_issue.fix_id`, and `queue.active_fix_ids`; fresh schemas include them.

stovepipe/extension/storage/mock/storage_mock.go

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "build_issue_store.go",
         "build_store.go",
+        "fix_store.go",
         "open_build_issue_store.go",
         "queue_log_store.go",
         "queue_store.go",
@@ -33,6 +34,7 @@ go_test(
     srcs = [
         "build_issue_store_test.go",
         "build_store_test.go",
+        "fix_store_test.go",
         "open_build_issue_store_test.go",
         "queue_store_test.go",
         "queue_summary_store_test.go",

stovepipe/extension/storage/mysql/BUILD.bazel

@@ -46,12 +46,13 @@ func (b *buildStore) Create(ctx context.Context, build entity.Build) (retErr err
 	defer func() { op.Complete(retErr) }()
 
 	_, err := b.db.ExecContext(ctx,
-		`INSERT INTO build (queue, id, request_id, status, version)
-		 VALUES (?, ?, ?, ?, ?)`,
+		`INSERT INTO build (queue, id, request_id, status, first_fix_id, version)
+		 VALUES (?, ?, ?, ?, ?, ?)`,
 		b.queue,
 		build.ID,
 		build.RequestID,
 		build.Status,
+		build.FirstFixID,
 		build.Version,
 	)
 	if err != nil {
@@ -71,13 +72,14 @@ func (b *buildStore) Get(ctx context.Context, id string) (ret entity.Build, retE
 
 	var build entity.Build
 	err := b.db.QueryRowContext(ctx,
-		`SELECT id, request_id, status, version
+		`SELECT id, request_id, status, first_fix_id, version
 		 FROM build WHERE queue = ? AND id = ?`,
 		b.queue, id,
 	).Scan(
 		&build.ID,
 		&build.RequestID,
 		&build.Status,
+		&build.FirstFixID,
 		&build.Version,
 	)
 
@@ -91,7 +93,7 @@ func (b *buildStore) Get(ctx context.Context, id string) (ret entity.Build, retE
 	return build, nil
 }
 
-// Update persists the mutable fields of build (status) if the stored version matches
+// Update persists the mutable build snapshot if the stored version matches
 // oldVersion, writing newVersion. Returns ErrVersionMismatch if the stored version does not
 // match (including when the build does not exist). This is a pure conditional write; the
 // caller owns version arithmetic.
@@ -101,9 +103,10 @@ func (b *buildStore) Update(ctx context.Context, build entity.Build, oldVersion,
 
 	result, err := b.db.ExecContext(ctx,
 		`UPDATE build
-		 SET status = ?, version = ?
+		 SET status = ?, first_fix_id = ?, version = ?
 		 WHERE queue = ? AND id = ? AND version = ?`,
 		build.Status,
+		build.FirstFixID,
 		newVersion,
 		b.queue,
 		build.ID,

stovepipe/extension/storage/mysql/build_store.go

@@ -58,15 +58,15 @@ func TestBuildStore_Create(t *testing.T) {
 			name: "success",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO build").
-					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version).
+					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.FirstFixID, build.Version).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
 		{
 			name: "duplicate id returns ErrAlreadyExists",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO build").
-					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version).
+					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.FirstFixID, build.Version).
 					WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry})
 			},
 			wantErr:   true,
@@ -76,7 +76,7 @@ func TestBuildStore_Create(t *testing.T) {
 			name: "other exec error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO build").
-					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version).
+					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.FirstFixID, build.Version).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -124,9 +124,9 @@ func TestBuildStore_Get(t *testing.T) {
 			name: "found",
 			id:   want.ID,
 			setup: func(mock sqlmock.Sqlmock) {
-				rows := sqlmock.NewRows([]string{"id", "request_id", "status", "version"}).
-					AddRow(want.ID, want.RequestID, string(want.Status), want.Version)
-				mock.ExpectQuery("SELECT id, request_id, status, version").
+				rows := sqlmock.NewRows([]string{"id", "request_id", "status", "first_fix_id", "version"}).
+					AddRow(want.ID, want.RequestID, string(want.Status), want.FirstFixID, want.Version)
+				mock.ExpectQuery("SELECT id, request_id, status, first_fix_id, version").
 					WithArgs("monorepo/main", want.ID).
 					WillReturnRows(rows)
 			},
@@ -136,7 +136,7 @@ func TestBuildStore_Get(t *testing.T) {
 			name: "not found",
 			id:   "missing",
 			setup: func(mock sqlmock.Sqlmock) {
-				mock.ExpectQuery("SELECT id, request_id, status, version").
+				mock.ExpectQuery("SELECT id, request_id, status, first_fix_id, version").
 					WithArgs("monorepo/main", "missing").
 					WillReturnError(sql.ErrNoRows)
 			},
@@ -147,7 +147,7 @@ func TestBuildStore_Get(t *testing.T) {
 			name: "query error",
 			id:   "bad",
 			setup: func(mock sqlmock.Sqlmock) {
-				mock.ExpectQuery("SELECT id, request_id, status, version").
+				mock.ExpectQuery("SELECT id, request_id, status, first_fix_id, version").
 					WithArgs("monorepo/main", "bad").
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
@@ -191,15 +191,15 @@ func TestBuildStore_Update(t *testing.T) {
 			name: "success",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE build").
-					WithArgs(build.Status, newVersion, "monorepo/main", build.ID, oldVersion).
+					WithArgs(build.Status, build.FirstFixID, newVersion, "monorepo/main", build.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
 		{
 			name: "version mismatch",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE build").
-					WithArgs(build.Status, newVersion, "monorepo/main", build.ID, oldVersion).
+					WithArgs(build.Status, build.FirstFixID, newVersion, "monorepo/main", build.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 0))
 			},
 			wantErr:   true,
@@ -209,7 +209,7 @@ func TestBuildStore_Update(t *testing.T) {
 			name: "exec error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE build").
-					WithArgs(build.Status, newVersion, "monorepo/main", build.ID, oldVersion).
+					WithArgs(build.Status, build.FirstFixID, newVersion, "monorepo/main", build.ID, oldVersion).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -218,7 +218,7 @@ func TestBuildStore_Update(t *testing.T) {
 			name: "rows affected error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE build").
-					WithArgs(build.Status, newVersion, "monorepo/main", build.ID, oldVersion).
+					WithArgs(build.Status, build.FirstFixID, newVersion, "monorepo/main", build.ID, oldVersion).
 					WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error")))
 			},
 			wantErr: true,

stovepipe/extension/storage/mysql/build_store_test.go

@@ -0,0 +1,172 @@
+// 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"
+	"encoding/json"
+	"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 fixStore struct {
+	db    *sql.DB
+	scope tally.Scope
+	queue string
+}
+
+// NewFixStore creates a MySQL-backed, queue-scoped FixStore.
+func NewFixStore(db *sql.DB, scope tally.Scope, queue string) storage.FixStore {
+	return &fixStore{db: db, scope: scope, queue: queue}
+}
+
+func (s *fixStore) Create(ctx context.Context, fix entity.Fix) (retErr error) {
+	op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
+
+	if fix.Queue != s.queue || fix.ID == "" {
+		return fmt.Errorf("repair requires a nonempty ID and queue matching %q", s.queue)
+	}
+	issuesJSON, metadataJSON, err := marshalFixJSON(fix)
+	if err != nil {
+		return fmt.Errorf("failed to marshal repair data: %w", err)
+	}
+
+	_, err = s.db.ExecContext(ctx,
+		`INSERT INTO fix (queue, id, build_id, state, issues, execution_ref, metadata, created_at, updated_at, version)
+		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		s.queue,
+		fix.ID,
+		fix.BuildID,
+		fix.State,
+		issuesJSON,
+		fix.ExecutionRef,
+		metadataJSON,
+		fix.CreatedAt,
+		fix.UpdatedAt,
+		fix.Version,
+	)
+	if err != nil {
+		if isDuplicateEntry(err) {
+			return fmt.Errorf("fix queue=%q: %w", s.queue, storage.ErrAlreadyExists)
+		}
+		return fmt.Errorf("failed to insert fix queue=%q: %w", s.queue, err)
+	}
+	return nil
+}
+
+func (s *fixStore) Get(ctx context.Context, id string) (ret entity.Fix, retErr error) {
+	op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
+
+	var fix entity.Fix
+	fix.Queue = s.queue
+	var issuesJSON []byte
+	var metadataJSON []byte
+	err := s.db.QueryRowContext(ctx,
+		`SELECT id, build_id, state, issues, execution_ref, metadata, created_at, updated_at, version FROM fix
+		 WHERE queue = ? AND id = ?`,
+		s.queue,
+		id,
+	).Scan(
+		&fix.ID,
+		&fix.BuildID,
+		&fix.State,
+		&issuesJSON,
+		&fix.ExecutionRef,
+		&metadataJSON,
+		&fix.CreatedAt,
+		&fix.UpdatedAt,
+		&fix.Version,
+	)
+	if errors.Is(err, sql.ErrNoRows) {
+		return entity.Fix{}, storage.WrapNotFound(err)
+	}
+	if err != nil {
+		return entity.Fix{}, fmt.Errorf("failed to get fix queue=%q: %w", s.queue, err)
+	}
+	if err := json.Unmarshal(issuesJSON, &fix.Issues); err != nil {
+		return entity.Fix{}, fmt.Errorf("failed to unmarshal repair issues: %w", err)
+	}
+	if err := json.Unmarshal(metadataJSON, &fix.Metadata); err != nil {
+		return entity.Fix{}, fmt.Errorf("failed to unmarshal repair metadata: %w", err)
+	}
+	if fix.Metadata == nil {
+		fix.Metadata = map[string]string{}
+	}
+	return fix, nil
+}
+
+func (s *fixStore) Update(ctx context.Context, fix entity.Fix, oldVersion, newVersion int32) (retErr error) {
+	op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets)
+	defer func() { op.Complete(retErr) }()
+
+	if fix.Queue != s.queue || fix.ID == "" {
+		return fmt.Errorf("repair requires a nonempty ID and queue matching %q", s.queue)
+	}
+	issuesJSON, metadataJSON, err := marshalFixJSON(fix)
+	if err != nil {
+		return fmt.Errorf("failed to marshal repair data: %w", err)
+	}
+
+	result, err := s.db.ExecContext(ctx,
+		`UPDATE fix SET build_id = ?, state = ?, issues = ?, execution_ref = ?, metadata = ?, created_at = ?, updated_at = ?, version = ?
+		 WHERE queue = ? AND id = ? AND version = ?`,
+		fix.BuildID,
+		fix.State,
+		issuesJSON,
+		fix.ExecutionRef,
+		metadataJSON,
+		fix.CreatedAt,
+		fix.UpdatedAt,
+		newVersion,
+		s.queue,
+		fix.ID,
+		oldVersion,
+	)
+	if err != nil {
+		return fmt.Errorf("failed to update fix queue=%q: %w", s.queue, err)
+	}
+	rowsAffected, err := result.RowsAffected()
+	if err != nil {
+		return fmt.Errorf("failed to get rows affected for fix queue=%q: %w", s.queue, err)
+	}
+	if rowsAffected != 1 {
+		return fmt.Errorf("version mismatch for fix queue=%q expected_version=%d: %w", s.queue, oldVersion, storage.ErrVersionMismatch)
+	}
+	return nil
+}
+
+func marshalFixJSON(fix entity.Fix) ([]byte, []byte, error) {
+	issuesJSON, err := json.Marshal(fix.Issues)
+	if err != nil {
+		return nil, nil, err
+	}
+	metadata := fix.Metadata
+	if metadata == nil {
+		metadata = map[string]string{}
+	}
+	metadataJSON, err := json.Marshal(metadata)
+	if err != nil {
+		return nil, nil, err
+	}
+	return issuesJSON, metadataJSON, nil
+}