feat(stovepipe): persist first-failure repair submissions

stovepipe/extension/storage/mysql/BUILD.bazel

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "build_issue_store.go",
         "build_store.go",
+        "fix_store.go",
         "open_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_issue_store_test.go",
         "queue_store_test.go",
         "queue_summary_store_test.go",

stovepipe/extension/storage/mysql/fix_store.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
+}

stovepipe/extension/storage/mysql/fix_store_test.go

@@ -0,0 +1,119 @@
+// 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"
+	"testing"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	driver "github.com/go-sql-driver/mysql"
+	"github.com/stretchr/testify/require"
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/storage"
+)
+
+func TestFixStoresErrors(t *testing.T) {
+	ctx := context.Background()
+	dbError := errors.New("database unavailable")
+	for _, scenario := range []string{"create duplicate", "create error", "get missing", "get error", "update error", "update stale", "rows affected error"} {
+		t.Run(scenario, func(t *testing.T) {
+			db, mock, err := sqlmock.New()
+			require.NoError(t, err)
+			defer db.Close()
+			backend, err := NewStorage(db, testMetrics())
+			require.NoError(t, err)
+			bound, err := backend.For("queue")
+			require.NoError(t, err)
+			store := bound.GetFixStore()
+			fix := entity.Fix{ID: "fix", Queue: "queue"}
+			var want error
+			switch scenario {
+			case "create duplicate":
+				mock.ExpectExec("INSERT INTO fix").WillReturnError(&driver.MySQLError{Number: mysqlErrDuplicateEntry})
+				want = storage.ErrAlreadyExists
+				err = store.Create(ctx, fix)
+			case "create error":
+				mock.ExpectExec("INSERT INTO fix").WillReturnError(dbError)
+				want = dbError
+				err = store.Create(ctx, fix)
+			case "get missing":
+				mock.ExpectQuery("SELECT").WillReturnError(sql.ErrNoRows)
+				want = storage.ErrNotFound
+				_, err = store.Get(ctx, fix.ID)
+			case "get error":
+				mock.ExpectQuery("SELECT").WillReturnError(dbError)
+				want = dbError
+				_, err = store.Get(ctx, fix.ID)
+			case "update error":
+				mock.ExpectExec("UPDATE fix").WillReturnError(dbError)
+				want = dbError
+				err = store.Update(ctx, fix, 1, 2)
+			case "update stale":
+				mock.ExpectExec("UPDATE fix").WillReturnResult(sqlmock.NewResult(0, 0))
+				want = storage.ErrVersionMismatch
+				err = store.Update(ctx, fix, 1, 2)
+			case "rows affected error":
+				mock.ExpectExec("UPDATE fix").WillReturnResult(sqlmock.NewErrorResult(dbError))
+				want = dbError
+				err = store.Update(ctx, fix, 1, 2)
+			}
+			require.ErrorIs(t, err, want)
+			require.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}
+
+func TestFixStoresRejectInvalidWrites(t *testing.T) {
+	ctx := context.Background()
+	for _, operation := range []string{"create", "update"} {
+		t.Run(operation, func(t *testing.T) {
+			db, mock, err := sqlmock.New()
+			require.NoError(t, err)
+			defer db.Close()
+			fixStore := NewFixStore(db, testMetrics(), "queue")
+			for _, fix := range []entity.Fix{{Queue: "queue"}, {ID: "fix", Queue: "other"}} {
+				if operation == "create" {
+					err = fixStore.Create(ctx, fix)
+				} else {
+					err = fixStore.Update(ctx, fix, 1, 2)
+				}
+				require.Error(t, err)
+			}
+			require.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}
+
+func TestFixStoresRejectCorruptCollections(t *testing.T) {
+	ctx := context.Background()
+	for _, tt := range []struct{ name, issues, metadata string }{
+		{"repair issue openings", `{}`, `{}`},
+		{"repair metadata", `[]`, `[]`},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			db, mock, err := sqlmock.New()
+			require.NoError(t, err)
+			defer db.Close()
+			mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows([]string{"id", "build_id", "state", "issues", "execution_ref", "metadata", "created_at", "updated_at", "version"}).
+				AddRow("fix", "build-1", "queued", tt.issues, "", tt.metadata, 1, 1, 1))
+			_, err = NewFixStore(db, testMetrics(), "queue").Get(ctx, "fix")
+			require.Error(t, err)
+			require.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}

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

@@ -0,0 +1,14 @@
+-- Repairs retain explicit lifecycle fields, typed issue opening membership, and optional metadata.
+CREATE TABLE IF NOT EXISTS fix (
+    queue         VARBINARY(1020) NOT NULL,
+    id            VARBINARY(1020) NOT NULL,
+    build_id      TEXT            NOT NULL,
+    state         VARCHAR(64)     NOT NULL,
+    issues        JSON            NOT NULL,
+    execution_ref TEXT            NOT NULL,
+    metadata      JSON            NOT NULL,
+    created_at    BIGINT          NOT NULL,
+    updated_at    BIGINT          NOT NULL,
+    version       INT             NOT NULL,
+    PRIMARY KEY (queue, id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

stovepipe/extension/storage/mysql/storage.go

@@ -58,6 +58,7 @@ func (s *Storage) For(queueName string) (storage.Storage, error) {
 		buildIssueStore:        NewBuildIssueStore(s.db, queueScope.SubScope("build_issue_store"), queueName),
 		openIssueStore:         NewOpenIssueStore(s.db, queueScope.SubScope("open_issue_store"), queueName),
 		validationFactStore:    NewValidationFactStore(s.db, queueScope.SubScope("validation_fact_store"), queueName),
+		fixStore:               NewFixStore(s.db, queueScope.SubScope("fix_store"), queueName),
 	}, nil
 }
 
@@ -80,6 +81,7 @@ type mysqlStorage struct {
 	buildIssueStore        storage.BuildIssueStore
 	openIssueStore         storage.OpenIssueStore
 	validationFactStore    storage.ValidationFactStore
+	fixStore               storage.FixStore
 }
 
 // Verify mysqlStorage implements the queue-scoped aggregate at compile time.
@@ -125,6 +127,10 @@ func (f *mysqlStorage) GetValidationFactStore() storage.ValidationFactStore {
 	return f.validationFactStore
 }
 
+func (f *mysqlStorage) GetFixStore() storage.FixStore {
+	return f.fixStore
+}
+
 func (f *mysqlStorage) GetQueueSummaryStore() storage.QueueSummaryStore { return f.queueSummaryStore }
 
 func (f *mysqlStorage) GetQueueLogStore() storage.QueueLogStore {

stovepipe/extension/storage/mysql/storage_test.go

@@ -153,6 +153,14 @@ func TestStorageForQueueMetrics(t *testing.T) {
 				return err
 			},
 		},
+		{
+			name: "fix_store",
+			op:   "get",
+			read: func(bound storage.Storage, _ string) error {
+				_, err := bound.GetFixStore().Get(context.Background(), "fix-id")
+				return err
+			},
+		},
 	}
 
 	for _, tt := range tests {

stovepipe/extension/storage/storage.go

@@ -102,4 +102,6 @@ type Storage interface {
 
 	// GetValidationFactStore returns the ValidationFactStore instance.
 	GetValidationFactStore() ValidationFactStore
+
+	GetFixStore() FixStore
 }