feat(stovepipe): store fixer submissions per issue and backend

stovepipe/entity/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "build_issue.go",
         "ingest.go",
         "issue_log.go",
+        "issue_repair.go",
         "issue_state.go",
         "issue_summary.go",
         "issue_tracking_state.go",

stovepipe/entity/issue_repair.go

@@ -0,0 +1,32 @@
+// 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
+
+// IssueRepair is one executor's repair submission state for an issue opening.
+// Empty FixID and SkipReason identify pending submission, not backend acceptance.
+type IssueRepair struct {
+	// Issue identifies the issue opening within a queue.
+	Issue OpenIssue
+	// ExecutorID is the stable queue-local executor identity.
+	ExecutorID string
+	// FixID is the backend request or run ID for accepted work, empty while pending or skipped.
+	FixID string
+	// SkipReason is the terminal skip reason, empty while pending or accepted.
+	SkipReason string
+	// Metadata contains executor-owned string values; nil and empty maps are equivalent.
+	Metadata map[string]string
+	// Version is the positive optimistic-lock version.
+	Version int32
+}

stovepipe/extension/storage/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "build_issue_store.go",
         "build_store.go",
         "issue_log_store.go",
+        "issue_repair_store.go",
         "issue_state_store.go",
         "issue_summary_store.go",
         "issue_tracking_store.go",

stovepipe/extension/storage/README.md

@@ -8,7 +8,7 @@ This is a separate contract from `submitqueue/extension/storage` — same shape
 
 ## Optimistic locking contract
 
-Entities that support concurrent mutation (`Request`, `Build`, `IssueState`, `IssueTrackingState`) carry an `int32 Version` field. `Update` methods take both `oldVersion` (the where-clause guard) and `newVersion` (the value to write) — the store performs a pure conditional write and never computes `oldVersion + 1` itself. Version arithmetic is owned by the controller: it computes `newVersion`, calls `Update`, and only assigns `entity.Version = newVersion` after the call succeeds. See [AGENTS.md](../../../AGENTS.md) and the [submitqueue storage README](../../../submitqueue/extension/storage/README.md#optimistic-locking-contract) for the full rationale and the caller pattern — the convention is identical here.
+Entities that support concurrent mutation (`Request`, `Build`, `IssueState`, `IssueTrackingState`, `IssueRepair`) carry an `int32 Version` field. `Update` methods take both `oldVersion` (the where-clause guard) and `newVersion` (the value to write) — the store performs a pure conditional write and never computes `oldVersion + 1` itself. Version arithmetic is owned by the controller: it computes `newVersion`, calls `Update`, and only assigns `entity.Version = newVersion` after the call succeeds. See [AGENTS.md](../../../AGENTS.md) and the [submitqueue storage README](../../../submitqueue/extension/storage/README.md#optimistic-locking-contract) for the full rationale and the caller pattern — the convention is identical here.
 
 ## Read-after-write consistency
 
@@ -36,9 +36,9 @@ Each observation is stored separately, not in a serialized collection. Observati
 
 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.
+`BuildIssue.FixID` is the legacy single-executor repair association. 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.
+`IssueRepairStore` retains versioned submission state keyed by the complete opening identity and executor ID. Pending rows have empty fix IDs and skip reasons. Create never overwrites existing state; Update replaces a snapshot guarded by its version. The controller creates pending rows and owns submission outcomes; executors own metadata and must preserve other fields when updating. Both writers reload on version conflicts before reapplying their edits. Fix IDs are not unique: an executor can accept multiple issues in one external request. Executor-owned metadata remains a `map[string]string` in Go and is serialized to the non-null JSON `metadata` column. Nil and empty maps are equivalent; empty metadata is stored as `{}`. String values may contain serialized JSON without interpretation by the store. Apply `mysql/schema/issue_repair.sql` before deploying the new readers and writers. Host adapters own backend-specific dispatch limits. The schema's `CREATE TABLE IF NOT EXISTS` does not alter existing tables.
 
 ## Issue state
 

stovepipe/extension/storage/issue_repair_store.go

@@ -0,0 +1,35 @@
+// 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=issue_repair_store.go -destination=mock/issue_repair_store_mock.go -package=mock
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/stovepipe/entity"
+)
+
+// IssueRepairStore retains versioned submissions keyed by opening and executor.
+// The controller owns FixID and SkipReason; executors own Metadata. Callers must
+// preserve other fields and reload on version conflicts before reapplying changes.
+type IssueRepairStore interface {
+	// Get returns ErrNotFound if no repair row exists.
+	Get(context.Context, entity.OpenIssue, string) (entity.IssueRepair, error)
+	// Create returns ErrAlreadyExists without overwriting existing state.
+	Create(context.Context, entity.IssueRepair) error
+	// Update replaces the record only at oldVersion; callers supply newVersion.
+	Update(ctx context.Context, repair entity.IssueRepair, oldVersion, newVersion int32) error
+}

stovepipe/extension/storage/mock/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "build_issue_store_mock.go",
         "build_store_mock.go",
         "issue_log_store_mock.go",
+        "issue_repair_store_mock.go",
         "issue_state_store_mock.go",
         "issue_summary_store_mock.go",
         "issue_tracking_store_mock.go",

stovepipe/extension/storage/mysql/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "build_issue_store.go",
         "build_store.go",
         "issue_log_store.go",
+        "issue_repair_store.go",
         "issue_state_store.go",
         "issue_summary_store.go",
         "issue_tracking_store.go",
@@ -41,6 +42,7 @@ go_test(
         "build_issue_update_test.go",
         "build_store_test.go",
         "issue_log_store_test.go",
+        "issue_repair_store_test.go",
         "issue_state_list_test.go",
         "issue_state_store_test.go",
         "issue_summary_store_test.go",

stovepipe/extension/storage/mysql/issue_repair_store.go

@@ -0,0 +1,137 @@
+// 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/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/storage"
+)
+
+type issueRepairStore struct {
+	db    *sql.DB
+	queue string
+}
+
+// NewIssueRepairStore binds repair submission state to a queue.
+func NewIssueRepairStore(db *sql.DB, queue string) storage.IssueRepairStore {
+	return &issueRepairStore{db: db, queue: queue}
+}
+
+func (s *issueRepairStore) validateKey(issue entity.OpenIssue, executorID string) error {
+	if s.queue == "" || issue.FirstSeenBuildID == "" || issue.TaskID == "" || executorID == "" || len(executorID) > 128 {
+		return fmt.Errorf("issue repair requires queue, opening, task, and executor IDs (executor at most 128 bytes)")
+	}
+	return validateBuildIssueKey(s.queue, issue.FirstSeenBuildID, issue.TaskID, issue.UnitID)
+}
+
+func (s *issueRepairStore) Get(ctx context.Context, issue entity.OpenIssue, executorID string) (entity.IssueRepair, error) {
+	if err := s.validateKey(issue, executorID); err != nil {
+		return entity.IssueRepair{}, err
+	}
+	result := entity.IssueRepair{Issue: issue, ExecutorID: executorID}
+	var metadataJSON []byte
+	err := s.db.QueryRowContext(ctx, `SELECT fix_id, skip_reason, metadata, version FROM issue_repair
+        WHERE queue=? AND opening_build_id=? AND task_id=? AND unit_id=? AND executor_id=?`,
+		s.queue, issue.FirstSeenBuildID, issue.TaskID, issue.UnitID, executorID).Scan(&result.FixID, &result.SkipReason, &metadataJSON, &result.Version)
+	if errors.Is(err, sql.ErrNoRows) {
+		return entity.IssueRepair{}, storage.WrapNotFound(err)
+	}
+	if err != nil {
+		return entity.IssueRepair{}, fmt.Errorf("read issue repair: %w", err)
+	}
+	if err := json.Unmarshal(metadataJSON, &result.Metadata); err != nil {
+		return entity.IssueRepair{}, fmt.Errorf("decode issue repair metadata: %w", err)
+	}
+	if result.Metadata == nil {
+		result.Metadata = map[string]string{}
+	}
+	return result, nil
+}
+
+func (s *issueRepairStore) Create(ctx context.Context, result entity.IssueRepair) error {
+	if err := s.validateRepair(result); err != nil {
+		return err
+	}
+	metadataJSON, err := marshalRepairMetadata(result.Metadata)
+	if err != nil {
+		return fmt.Errorf("encode issue repair metadata: %w", err)
+	}
+	_, err = s.db.ExecContext(ctx, `INSERT INTO issue_repair
+		(queue,opening_build_id,task_id,unit_id,executor_id,fix_id,skip_reason,metadata,version) VALUES (?,?,?,?,?,?,?,?,?)`,
+		s.queue, result.Issue.FirstSeenBuildID, result.Issue.TaskID, result.Issue.UnitID, result.ExecutorID, result.FixID, result.SkipReason, metadataJSON, result.Version)
+	if isDuplicateEntry(err) {
+		return storage.ErrAlreadyExists
+	}
+	if err != nil {
+		return fmt.Errorf("create issue repair: %w", err)
+	}
+	return nil
+}
+
+func (s *issueRepairStore) Update(ctx context.Context, repair entity.IssueRepair, oldVersion, newVersion int32) error {
+	replacement := repair
+	replacement.Version = newVersion
+	if err := s.validateRepair(replacement); err != nil {
+		return err
+	}
+	if oldVersion <= 0 {
+		return fmt.Errorf("issue repair guard version must be positive")
+	}
+	metadataJSON, err := marshalRepairMetadata(repair.Metadata)
+	if err != nil {
+		return err
+	}
+	result, err := s.db.ExecContext(ctx, `UPDATE issue_repair SET fix_id=?, skip_reason=?, metadata=?, version=?
+        WHERE queue=? AND opening_build_id=? AND task_id=? AND unit_id=? AND executor_id=? AND version=?`,
+		repair.FixID, repair.SkipReason, metadataJSON, newVersion,
+		s.queue, repair.Issue.FirstSeenBuildID, repair.Issue.TaskID, repair.Issue.UnitID, repair.ExecutorID, oldVersion)
+	if err != nil {
+		return fmt.Errorf("update issue repair: %w", err)
+	}
+	rows, err := result.RowsAffected()
+	if err != nil {
+		return fmt.Errorf("read issue repair update count: %w", err)
+	}
+	if rows != 1 {
+		return storage.ErrVersionMismatch
+	}
+	return nil
+}
+
+func (s *issueRepairStore) validateRepair(repair entity.IssueRepair) error {
+	if err := s.validateKey(repair.Issue, repair.ExecutorID); err != nil {
+		return err
+	}
+	if repair.FixID != "" && repair.SkipReason != "" || len(repair.FixID) > 255 || len(repair.SkipReason) > 255 {
+		return fmt.Errorf("issue repair permits at most one fix ID or skip reason (at most 255 bytes)")
+	}
+	if repair.Version <= 0 {
+		return fmt.Errorf("issue repair version must be positive")
+	}
+	return nil
+}
+
+func marshalRepairMetadata(metadata map[string]string) ([]byte, error) {
+	if metadata == nil {
+		metadata = map[string]string{}
+	}
+	return json.Marshal(metadata)
+}