feat(stovepipe): select and submit one repair per build

stovepipe/controller/fix/BUILD.bazel

@@ -0,0 +1,38 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["fix.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/controller/fix",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//platform/consumer:go_default_library",
+        "//platform/metrics:go_default_library",
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/fixer:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["fix_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/consumer:go_default_library",
+        "//platform/errs:go_default_library",
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/fixer:go_default_library",
+        "//stovepipe/extension/fixer/mock:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "//stovepipe/extension/storage/mock:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+        "@org_uber_go_zap//zaptest/observer:go_default_library",
+    ],
+)

stovepipe/controller/fix/README.md

@@ -0,0 +1,7 @@
+# Fix submission
+
+The controller pages through a build's durable issue records and skips submission when any record has a fix ID, including recovered issues. Otherwise, it selects one currently open issue whose first-seen build matches this build and calls `RepairExecutor.Submit`.
+
+A returned fix ID is associated with the issue through `BuildIssueStore.Update`, using the version read during selection. A concurrent association is preserved. A submission error without an ID leaves the issue eligible for normal queue retries; an ID returned alongside an error is retained.
+
+The one-per-build limit is controller policy. Submission precedes persistence, so overlapping deliveries, crashes, and failed writes can still produce duplicate executor calls. Issue recovery remains separate from repair association.

stovepipe/controller/fix/fix.go

@@ -0,0 +1,162 @@
+// 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 fix submits build-scoped repair work.
+package fix
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/fixer"
+	"github.com/uber/submitqueue/stovepipe/extension/storage"
+	"go.uber.org/zap"
+)
+
+const buildIssuePageSize = 100
+
+type Controller struct {
+	logger        *zap.SugaredLogger
+	metricsScope  tally.Scope
+	stores        storage.Factory
+	executors     fixer.Factory
+	topicKey      consumer.TopicKey
+	consumerGroup string
+}
+
+func NewController(
+	logger *zap.SugaredLogger,
+	scope tally.Scope,
+	stores storage.Factory,
+	executors fixer.Factory,
+	topicKey consumer.TopicKey,
+	consumerGroup string,
+) *Controller {
+	return &Controller{
+		logger:        logger.Named("fix_controller"),
+		metricsScope:  scope.SubScope("fix_controller"),
+		stores:        stores,
+		executors:     executors,
+		topicKey:      topicKey,
+		consumerGroup: consumerGroup,
+	}
+}
+
+func (c *Controller) processBuild(ctx context.Context, queueName, buildID string) error {
+	store, err := c.stores.For(storage.Config{QueueName: queueName})
+	if err != nil {
+		return fmt.Errorf("failed to resolve repair storage: %w", err)
+	}
+	observation, found, err := findIssueToFix(ctx, store.GetBuildIssueStore(), store.GetOpenIssueStore(), buildID)
+	if err != nil {
+		return fmt.Errorf("failed to select an issue for build %q: %w", buildID, err)
+	}
+	if !found {
+		return nil
+	}
+
+	executor, err := c.executors.For(fixer.Config{QueueName: queueName})
+	if err != nil {
+		return fmt.Errorf("failed to resolve repair executor: %w", err)
+	}
+
+	op := metrics.Begin(c.metricsScope, "submit", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...)
+	issue := entity.OpenIssue{TaskID: observation.TaskID, UnitID: observation.UnitID, FirstSeenBuildID: buildID}
+	fixID, err := executor.Submit(ctx, issue)
+	if err == nil && fixID == "" {
+		err = fmt.Errorf("repair executor returned an empty fix ID")
+	}
+	op.Complete(err)
+	if err != nil {
+		c.logger.Errorw("repair submission failed",
+			"queue", queueName, "build_id", buildID,
+			"task_id", issue.TaskID, "unit_id", issue.UnitID,
+			"fix_id", fixID, "error", err)
+		if fixID == "" {
+			return fmt.Errorf("failed to submit repair for build %q: %w", buildID, err)
+		}
+	} else {
+		c.logger.Infow("repair submitted",
+			"queue", queueName, "build_id", buildID,
+			"task_id", issue.TaskID, "unit_id", issue.UnitID, "fix_id", fixID)
+	}
+
+	// Submission precedes persistence; a crash or write failure can allow duplicates.
+	updated := observation
+	updated.FixID = fixID
+	observations := store.GetBuildIssueStore()
+	if err := observations.Update(ctx, buildID, updated, observation.Version, observation.Version+1); err != nil {
+		if !errors.Is(err, storage.ErrVersionMismatch) {
+			return fmt.Errorf("failed to associate repair with build issue: %w", err)
+		}
+		current, readErr := observations.Get(ctx, buildID, observation.TaskID, observation.UnitID)
+		if readErr != nil {
+			return fmt.Errorf("failed to reload build issue after concurrent update: %w", readErr)
+		}
+		if current.FixID == "" {
+			return err
+		}
+		c.logger.Warnw("preserving previously recorded repair association",
+			"queue", queueName, "build_id", buildID,
+			"task_id", issue.TaskID, "unit_id", issue.UnitID,
+			"fix_id", current.FixID, "submitted_fix_id", fixID)
+	}
+	return nil
+}
+
+func (c *Controller) Name() string { return "fix" }
+
+func (c *Controller) TopicKey() consumer.TopicKey { return c.topicKey }
+
+func (c *Controller) ConsumerGroup() string { return c.consumerGroup }
+
+func findIssueToFix(ctx context.Context, observations storage.BuildIssueStore, openIssues storage.OpenIssueStore, buildID string) (entity.BuildIssue, bool, error) {
+	bounds := storage.BuildIssueRange{Limit: buildIssuePageSize}
+	var candidate entity.BuildIssue
+	found := false
+	for {
+		page, err := observations.List(ctx, buildID, bounds)
+		if err != nil {
+			return entity.BuildIssue{}, false, err
+		}
+		if len(page) == 0 {
+			return candidate, found, nil
+		}
+		for _, record := range page {
+			if record.FixID != "" {
+				return entity.BuildIssue{}, false, nil
+			}
+			if found {
+				continue
+			}
+			issue, err := openIssues.Get(ctx, record.TaskID, record.UnitID)
+			if errors.Is(err, storage.ErrNotFound) {
+				continue
+			}
+			if err != nil {
+				return entity.BuildIssue{}, false, err
+			}
+			if issue.FirstSeenBuildID == buildID {
+				candidate, found = record, true
+			}
+		}
+		last := page[len(page)-1]
+		bounds.After = storage.BuildIssueCursor{TaskID: last.TaskID, UnitID: last.UnitID}
+	}
+}