feat(stovepipe): extend fixer API for batching and multiple backends

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/fixer/fixer.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// Package fixer defines repair submission for an open issue.
+// Package fixer defines repair submission for open issues.
 package fixer
 
 //go:generate mockgen -source=fixer.go -destination=mock/fixer_mock.go -package=mock
@@ -23,22 +23,42 @@ import (
 	"github.com/uber/submitqueue/stovepipe/entity"
 )
 
-// RepairExecutor submits repair work for one open issue within a configured queue.
-// Implementations must be concurrency-safe and return promptly after submission.
+// RepairExecutor chooses how to group repair work. Implementations must be concurrency-safe.
 type RepairExecutor interface {
-	// Submit returns a nonempty opaque fix ID on acceptance, without establishing recovery.
-	// An ID returned alongside an error still identifies the repair attempt.
-	// Errors without an ID may follow acceptance; repeated calls can start additional work.
-	Submit(context.Context, entity.OpenIssue) (string, error)
+	// Submit returns per-item backend acceptance or terminal skips, not repair completion.
+	// Valid submissions returned alongside errors are persisted; omitted items remain pending.
+	// Retries may include subsets or additional items; implementations deduplicate by opening.
+	// Pending rows exist before Submit. Executors manage metadata through injected storage,
+	// preserving submission fields and reloading on version conflicts.
+	Submit(context.Context, []entity.OpenIssue) ([]Submission, error)
 }
 
-// Config identifies the queue whose repair executor is requested.
+// Submission associates one input item with accepted repair work or a deliberate skip.
+// Acceptance does not indicate repair completion or success.
+type Submission struct {
+	// Issue identifies the input opening, including its first-seen build.
+	Issue entity.OpenIssue
+	// FixID is the backend request or run ID for accepted work; multiple items may share it.
+	FixID string
+	// SkipReason is nonempty for a terminal skip. Exactly one of FixID and SkipReason is required.
+	SkipReason string
+}
+
+// Registration binds a stable queue-local identity to an executor.
+type Registration struct {
+	// ID is durable across configuration changes and must not be reused for another backend.
+	ID string
+	// Executor implements the registered repair policy.
+	Executor RepairExecutor
+}
+
+// Config identifies a validation queue.
 type Config struct {
 	// QueueName identifies the validation queue.
 	QueueName string
 }
 
-// Factory resolves queue-bound executors; routing belongs to service wiring.
+// Factory resolves all registered executors; an empty slice disables dispatch.
 type Factory interface {
-	For(Config) (RepairExecutor, error)
+	For(Config) ([]Registration, error)
 }

stovepipe/extension/fixer/noop/noop.go

@@ -29,6 +29,10 @@ func New() fixer.RepairExecutor {
 	return executor{}
 }
 
-func (executor) Submit(_ context.Context, issue entity.OpenIssue) (string, error) {
-	return "noop/" + issue.FirstSeenBuildID, nil
+func (executor) Submit(_ context.Context, issues []entity.OpenIssue) ([]fixer.Submission, error) {
+	submissions := make([]fixer.Submission, 0, len(issues))
+	for _, issue := range issues {
+		submissions = append(submissions, fixer.Submission{Issue: issue, FixID: "noop/" + issue.FirstSeenBuildID})
+	}
+	return submissions, nil
 }

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",