feat(stovepipe): Reconcile issues in batches

Author: prathshenoy

stovepipe/controller/issuetrack/BUILD.bazel

@@ -0,0 +1,35 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["issuetrack.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/controller/issuetrack",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//platform/base/id:go_default_library",
+        "//platform/consumer:go_default_library",
+        "//platform/metrics:go_default_library",
+        "//stovepipe/core/loader:go_default_library",
+        "//stovepipe/entity: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 = ["issuetrack_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/entity: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",
+    ],
+)

stovepipe/controller/issuetrack/README.md

@@ -0,0 +1,13 @@
+# Issue reconciliation
+
+The controller's batch operation advances one persisted issue-tracking pass by at most 100 distinct task/unit keys. A refresh scans the source build's observations and opens missing issues. Finalization merges that ordered range with the queue's ordered open-issue range: observed keys are kept or opened, and open-only keys are resolved. Both ranges use the same exclusive cursor, so no separate opening/resolution scan state is required. Cancelled builds only add observations.
+
+Each read is bounded. When a page fills the limit, another batch may be needed to establish that the range is exhausted. After all writes succeed, the controller conditionally replaces progress with the last fully processed key and incremented version. Exhausted passes become refreshed or finalized. Failure before checkpoint persistence leaves the old cursor available for replay; opening/history writes and resolution-before-deletion are idempotent. Each pass reloads its persisted source and final passes require the request's winning terminal build.
+
+Refresh scans are not snapshots: observations inserted behind a cursor are discovered by a later refresh starting from the beginning. Finalization requires successful complete detection and persistence before its handoff; terminal build status alone does not establish completeness. Final reconciliation uses the build's accumulated issue set, not merely its latest detector response.
+
+This layer does not receive messages, publish repair requests, register consumers, or implement terminal recovery. Delivery orchestration is added separately. A future repair handoff must persist its prerequisites and publish before advancing the checkpoint past that obligation.
+
+## Ownership prerequisite
+
+All issue membership and history writes require one active writer for a queue. A progress CAS is not a lock or a fence for those other records. In particular, a worker paused before creating an opening can resume after another worker finalized a newer build, recreate the old opening, and only then lose its checkpoint CAS. Normal per-partition serialization does not by itself prove safety during lease transfer. Worker fencing must be established before enabling this controller in a host; checkpointing does not resolve that rollout blocker.

stovepipe/controller/issuetrack/issuetrack.go

@@ -0,0 +1,260 @@
+// 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 issuetrack applies build observations to operational open issues and lifecycle history.
+// All deliveries for one queue must share its partition and consumer group.
+package issuetrack
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/platform/base/id"
+	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/stovepipe/core/loader"
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/storage"
+	"go.uber.org/zap"
+)
+
+// Each batch applies at most this many distinct issue keys.
+const _issueBatchSize = 100
+
+// Controller is the queue-serialized writer of open issues and issue transitions.
+// Accepted pending work is replayed before a newer source can change issue state.
+type Controller struct {
+	logger        *zap.SugaredLogger
+	metricsScope  tally.Scope
+	stores        storage.Factory
+	topicKey      consumer.TopicKey
+	consumerGroup string
+	nowMs         func() int64
+}
+
+func NewController(
+	logger *zap.SugaredLogger,
+	scope tally.Scope,
+	stores storage.Factory,
+	topicKey consumer.TopicKey,
+	consumerGroup string,
+) *Controller {
+	return &Controller{
+		logger:        logger.Named("issuetrack_controller"),
+		metricsScope:  scope.SubScope("issuetrack_controller"),
+		stores:        stores,
+		topicKey:      topicKey,
+		consumerGroup: consumerGroup,
+		nowMs:         func() int64 { return time.Now().UnixMilli() },
+	}
+}
+
+func (c *Controller) applyReconciliationBatch(ctx context.Context, store storage.Storage, queue string, progress entity.IssueTrackingState) (checkpoint entity.IssueTrackingState, retErr error) {
+	op := metrics.Begin(c.metricsScope, "batch", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...)
+	defer func() { op.Complete(retErr) }()
+
+	build, request, err := loadBuildAndRequest(ctx, store, queue, progress.BuildID)
+	if err != nil {
+		return entity.IssueTrackingState{}, err
+	}
+	if request.ID != progress.RequestID || !progress.IsPending() {
+		return entity.IssueTrackingState{}, fmt.Errorf("issue batch requires pending work for its source request")
+	}
+	finalizing := progress.Phase == entity.IssueTrackingPhaseFinalizing
+	if finalizing {
+		if err := validateFinalBuild(build, request); err != nil {
+			return entity.IssueTrackingState{}, err
+		}
+	}
+
+	observations, err := store.GetBuildIssueStore().List(ctx, build.ID, storage.BuildIssueRange{
+		After: storage.BuildIssueCursor{TaskID: progress.AfterTaskID, UnitID: progress.AfterUnitID},
+		Limit: _issueBatchSize,
+	})
+	if err != nil {
+		return entity.IssueTrackingState{}, fmt.Errorf("list build observations: %w", err)
+	}
+
+	var openIssues []entity.OpenIssue
+	if finalizing && build.Status != entity.BuildStatusCancelled {
+		openIssues, err = store.GetOpenIssueStore().List(ctx, storage.OpenIssueRange{
+			After: storage.OpenIssueCursor{TaskID: progress.AfterTaskID, UnitID: progress.AfterUnitID},
+			Limit: _issueBatchSize,
+		})
+		if err != nil {
+			return entity.IssueTrackingState{}, fmt.Errorf("list open issues: %w", err)
+		}
+	}
+
+	next := progress
+	observedIndex, openIndex := 0, 0
+	for processed := 0; processed < _issueBatchSize && (observedIndex < len(observations) || openIndex < len(openIssues)); processed++ {
+		if err := ctx.Err(); err != nil {
+			return entity.IssueTrackingState{}, err
+		}
+
+		useObservation := openIndex == len(openIssues)
+		if observedIndex < len(observations) && openIndex < len(openIssues) {
+			observed, open := observations[observedIndex], openIssues[openIndex]
+			useObservation = compareIssueKeys(observed.TaskID, observed.UnitID, open.TaskID, open.UnitID) <= 0
+		}
+		if useObservation {
+			issue := observations[observedIndex]
+			if err := c.openObservedIssue(ctx, store, build.ID, issue); err != nil {
+				return entity.IssueTrackingState{}, err
+			}
+			if openIndex < len(openIssues) && openIssues[openIndex].TaskID == issue.TaskID && openIssues[openIndex].UnitID == issue.UnitID {
+				openIndex++
+			}
+			observedIndex++
+			next.AfterTaskID, next.AfterUnitID = issue.TaskID, issue.UnitID
+		} else {
+			issue := openIssues[openIndex]
+			if err := c.resolveIssueIfAbsent(ctx, store, build.ID, issue); err != nil {
+				return entity.IssueTrackingState{}, err
+			}
+			openIndex++
+			next.AfterTaskID, next.AfterUnitID = issue.TaskID, issue.UnitID
+		}
+	}
+
+	// Full pages may hide more keys; an additional empty read safely establishes completion.
+	complete := observedIndex == len(observations) && openIndex == len(openIssues) &&
+		len(observations) < _issueBatchSize && len(openIssues) < _issueBatchSize
+	if complete {
+		next.Phase = entity.IssueTrackingPhaseRefreshed
+		if finalizing {
+			next.Phase = entity.IssueTrackingPhaseFinalized
+		}
+	}
+
+	newVersion := progress.Version + 1
+	if err := store.GetIssueTrackingStore().Update(ctx, next, progress.Version, newVersion); err != nil {
+		return entity.IssueTrackingState{}, fmt.Errorf("checkpoint issue reconciliation: %w", err)
+	}
+	next.Version = newVersion
+	return next, nil
+}
+
+func compareIssueKeys(taskID, unitID, otherTaskID, otherUnitID string) int {
+	if taskID < otherTaskID {
+		return -1
+	}
+	if taskID > otherTaskID {
+		return 1
+	}
+	if unitID < otherUnitID {
+		return -1
+	}
+	if unitID > otherUnitID {
+		return 1
+	}
+	return 0
+}
+
+func (c *Controller) openObservedIssue(ctx context.Context, store storage.Storage, buildID string, issue entity.BuildIssue) error {
+	openStore := store.GetOpenIssueStore()
+	openIssue, err := openStore.Get(ctx, issue.TaskID, issue.UnitID)
+	if err != nil && !errors.Is(err, storage.ErrNotFound) {
+		return fmt.Errorf("read open issue: %w", err)
+	}
+
+	if errors.Is(err, storage.ErrNotFound) {
+		openIssue = entity.OpenIssue{
+			TaskID:           issue.TaskID,
+			UnitID:           issue.UnitID,
+			FirstSeenBuildID: buildID,
+		}
+		err = openStore.Create(ctx, openIssue)
+		if errors.Is(err, storage.ErrAlreadyExists) {
+			openIssue, err = openStore.Get(ctx, issue.TaskID, issue.UnitID)
+		}
+		if err != nil {
+			return fmt.Errorf("open observed issue: %w", err)
+		}
+	}
+
+	if openIssue.FirstSeenBuildID == buildID {
+		return c.recordIssueEvent(ctx, store, buildID, entity.IssueEventOpened, issue.TaskID, issue.UnitID)
+	}
+	return nil
+}
+
+func (c *Controller) resolveIssueIfAbsent(ctx context.Context, store storage.Storage, buildID string, issue entity.OpenIssue) error {
+	_, err := store.GetBuildIssueStore().Get(ctx, buildID, issue.TaskID, issue.UnitID)
+	if err == nil {
+		return nil
+	}
+	if !errors.Is(err, storage.ErrNotFound) {
+		return fmt.Errorf("check retained build observation: %w", err)
+	}
+
+	// Record resolution before deletion so a failed history write leaves the issue retryable.
+	if err := c.recordIssueEvent(ctx, store, buildID, entity.IssueEventResolved, issue.TaskID, issue.UnitID); err != nil {
+		return err
+	}
+	if err := store.GetOpenIssueStore().Delete(ctx, issue); err != nil {
+		return fmt.Errorf("resolve open issue: %w", err)
+	}
+	return nil
+}
+
+func (c *Controller) recordIssueEvent(ctx context.Context, store storage.Storage, buildID string, event entity.IssueEvent, taskID, unitID string) error {
+	log := entity.IssueLog{
+		TaskID:      taskID,
+		UnitID:      unitID,
+		BuildID:     buildID,
+		Event:       event,
+		TimestampMs: c.nowMs(),
+	}
+	if err := store.GetIssueLogStore().Create(ctx, log); err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
+		return fmt.Errorf("record issue transition: %w", err)
+	}
+	return nil
+}
+
+func loadBuildAndRequest(ctx context.Context, store storage.Storage, queue, buildID string) (entity.Build, entity.Request, error) {
+	build, err := loader.ByID(ctx, buildID, store.GetBuildStore().Get, "build")
+	if err != nil {
+		return entity.Build{}, entity.Request{}, err
+	}
+
+	request, err := loader.ByID(ctx, build.RequestID, store.GetRequestStore().Get, "request")
+	if err != nil {
+		return entity.Build{}, entity.Request{}, err
+	}
+
+	if request.Queue != queue {
+		return entity.Build{}, entity.Request{}, fmt.Errorf("issue tracking request belongs to another queue")
+	}
+	if err := id.Validate(request.ID); err != nil {
+		return entity.Build{}, entity.Request{}, fmt.Errorf("invalid issue tracking request ID: %w", err)
+	}
+
+	return build, request, nil
+}
+
+func validateFinalBuild(build entity.Build, request entity.Request) error {
+	if !build.Status.IsTerminal() || request.TerminalBuildID != build.ID {
+		return fmt.Errorf("final issue reconciliation requires the request's terminal build")
+	}
+	return nil
+}
+
+func (c *Controller) Name() string                { return "issuetrack" }
+func (c *Controller) TopicKey() consumer.TopicKey { return c.topicKey }
+func (c *Controller) ConsumerGroup() string       { return c.consumerGroup }