feat(stovepipe): Save planned issue changes

stovepipe/controller/fix/fix.go

@@ -63,7 +63,7 @@ func (c *Controller) processBuild(ctx context.Context, queueName, buildID string
 	if err != nil {
 		return fmt.Errorf("failed to resolve repair storage: %w", err)
 	}
-	observation, found, err := findIssueToFix(ctx, store.GetBuildIssueStore(), store.GetOpenIssueStore(), buildID)
+	observation, found, err := findIssueToFix(ctx, store.GetBuildIssueStore(), store.GetIssueStateStore(), buildID)
 	if err != nil {
 		return fmt.Errorf("failed to select an issue for build %q: %w", buildID, err)
 	}
@@ -126,7 +126,7 @@ 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) {
+func findIssueToFix(ctx context.Context, observations storage.BuildIssueStore, issues storage.IssueStateStore, buildID string) (entity.BuildIssue, bool, error) {
 	bounds := storage.BuildIssueRange{Limit: buildIssuePageSize}
 	var candidate entity.BuildIssue
 	found := false
@@ -145,14 +145,14 @@ func findIssueToFix(ctx context.Context, observations storage.BuildIssueStore, o
 			if found {
 				continue
 			}
-			issue, err := openIssues.Get(ctx, record.TaskID, record.UnitID)
+			issue, err := issues.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 {
+			if issue.Status == entity.IssueStatusOpen && issue.FirstSeenBuildID == buildID {
 				candidate, found = record, true
 			}
 		}

stovepipe/controller/fix/fix_test.go

@@ -42,7 +42,7 @@ type controllerMocks struct {
 	ctrl         *gomock.Controller
 	stores       *storagemock.MockFactory
 	store        *storagemock.MockStorage
-	issues       *storagemock.MockOpenIssueStore
+	issues       *storagemock.MockIssueStateStore
 	observations *storagemock.MockBuildIssueStore
 	executors    *fixermock.MockFactory
 	executor     *fixermock.MockRepairExecutor
@@ -57,13 +57,13 @@ func newTestController(t *testing.T) (*Controller, controllerMocks) {
 		ctrl:         ctrl,
 		stores:       storagemock.NewMockFactory(ctrl),
 		store:        storagemock.NewMockStorage(ctrl),
-		issues:       storagemock.NewMockOpenIssueStore(ctrl),
+		issues:       storagemock.NewMockIssueStateStore(ctrl),
 		observations: storagemock.NewMockBuildIssueStore(ctrl),
 		executors:    fixermock.NewMockFactory(ctrl),
 		executor:     fixermock.NewMockRepairExecutor(ctrl),
 		logs:         logs,
 	}
-	m.store.EXPECT().GetOpenIssueStore().Return(m.issues).AnyTimes()
+	m.store.EXPECT().GetIssueStateStore().Return(m.issues).AnyTimes()
 	m.store.EXPECT().GetBuildIssueStore().Return(m.observations).AnyTimes()
 	return NewController(zap.New(core).Sugar(), tally.NoopScope, m.stores, m.executors,
 		consumer.TopicKey("fix"), "stovepipe-fix"), m
@@ -73,6 +73,10 @@ func testIssue() entity.OpenIssue {
 	return entity.OpenIssue{TaskID: "test", UnitID: "unit/1 ", FirstSeenBuildID: testBuildID}
 }
 
+func storedIssue(issue entity.OpenIssue) entity.IssueState {
+	return entity.IssueState{TaskID: issue.TaskID, UnitID: issue.UnitID, FirstSeenBuildID: issue.FirstSeenBuildID, Status: entity.IssueStatusOpen, Version: 1}
+}
+
 func issueIdentity(issue entity.OpenIssue) entity.BuildIssue {
 	return entity.BuildIssue{TaskID: issue.TaskID, UnitID: issue.UnitID}
 }
@@ -94,7 +98,7 @@ func (m controllerMocks) expectIssue(issue entity.OpenIssue) {
 	m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil)
 	m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}).
 		Return([]entity.BuildIssue{issueIdentity(issue)}, nil)
-	m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(issue, nil)
+	m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(storedIssue(issue), nil)
 	m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue))).Return(nil, nil)
 	m.executors.EXPECT().For(fixer.Config{QueueName: testQueue}).Return(m.executor, nil)
 }
@@ -108,7 +112,7 @@ func TestProcessBuildSubmitsOneIssueAndAssociatesFixAfterCall(t *testing.T) {
 	m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil)
 	m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}).
 		Return([]entity.BuildIssue{observation, another}, nil)
-	m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(issue, nil)
+	m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(storedIssue(issue), nil)
 	m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(another)).Return(nil, nil)
 	m.executors.EXPECT().For(fixer.Config{QueueName: testQueue}).Return(m.executor, nil)
 	updated := observation
@@ -136,10 +140,10 @@ func TestProcessBuildFindsCandidateOnLaterPage(t *testing.T) {
 		m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}).
 			Return([]entity.BuildIssue{older}, nil),
 		m.issues.EXPECT().Get(gomock.Any(), older.TaskID, older.UnitID).
-			Return(entity.OpenIssue{TaskID: older.TaskID, FirstSeenBuildID: "older-build"}, nil),
+			Return(entity.IssueState{TaskID: older.TaskID, FirstSeenBuildID: "older-build", Status: entity.IssueStatusOpen, Version: 1}, nil),
 		m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(older)).
 			Return([]entity.BuildIssue{issueIdentity(issue)}, nil),
-		m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(issue, nil),
+		m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(storedIssue(issue), nil),
 		m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue))).Return(nil, nil),
 	)
 	m.executors.EXPECT().For(fixer.Config{QueueName: testQueue}).Return(m.executor, nil)
@@ -156,7 +160,7 @@ func TestProcessBuildChecksLaterPagesForRecordedFix(t *testing.T) {
 	gomock.InOrder(
 		m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}).
 			Return([]entity.BuildIssue{issueIdentity(issue)}, nil),
-		m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(issue, nil),
+		m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(storedIssue(issue), nil),
 		m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue))).
 			Return([]entity.BuildIssue{{TaskID: "resolved", FixID: "fix/1"}}, nil),
 	)
@@ -172,7 +176,7 @@ func TestProcessBuildDoesNotResubmitAfterRecordedFixID(t *testing.T) {
 		DoAndReturn(func(context.Context, string, storage.BuildIssueRange) ([]entity.BuildIssue, error) {
 			return []entity.BuildIssue{record}, nil
 		}).Times(2)
-	m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(issue, nil)
+	m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(storedIssue(issue), nil)
 	m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(record)).Return(nil, nil)
 	m.executors.EXPECT().For(fixer.Config{QueueName: testQueue}).Return(m.executor, nil)
 	m.executor.EXPECT().Submit(gomock.Any(), issue).Return("fix/1", nil)
@@ -196,10 +200,10 @@ func TestProcessBuildWithoutEligibleIssue(t *testing.T) {
 			m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}).
 				Return([]entity.BuildIssue{issue}, nil)
 			if firstSeenBuildID == "" {
-				m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(entity.OpenIssue{}, storage.ErrNotFound)
+				m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(entity.IssueState{}, storage.ErrNotFound)
 			} else {
 				m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).
-					Return(entity.OpenIssue{TaskID: issue.TaskID, FirstSeenBuildID: firstSeenBuildID}, nil)
+					Return(entity.IssueState{TaskID: issue.TaskID, FirstSeenBuildID: firstSeenBuildID, Status: entity.IssueStatusOpen, Version: 1}, nil)
 			}
 			m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issue)).Return(nil, nil)
 
@@ -303,10 +307,10 @@ func TestProcessBuildReturnsErrorsBeforeSubmission(t *testing.T) {
 				first.Return([]entity.BuildIssue{issueIdentity(issue)}, nil)
 				lookup := m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID)
 				if step == "open issue lookup" {
-					lookup.Return(entity.OpenIssue{}, assert.AnError)
+					lookup.Return(entity.IssueState{}, assert.AnError)
 					return
 				}
-				lookup.Return(issue, nil)
+				lookup.Return(storedIssue(issue), nil)
 				last := m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue)))
 				if step == "later page" {
 					last.Return(nil, assert.AnError)
@@ -320,3 +324,15 @@ func TestProcessBuildReturnsErrorsBeforeSubmission(t *testing.T) {
 		})
 	}
 }
+
+func TestProcessBuildSkipsClosedIssue(t *testing.T) {
+	controller, m := newTestController(t)
+	issue := testIssue()
+	closed := storedIssue(issue)
+	closed.Status = entity.IssueStatusClosed
+	m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil)
+	m.observations.EXPECT().List(gomock.Any(), testBuildID, storage.BuildIssueRange{Limit: buildIssuePageSize}).Return([]entity.BuildIssue{issueIdentity(issue)}, nil)
+	m.issues.EXPECT().Get(gomock.Any(), issue.TaskID, issue.UnitID).Return(closed, nil)
+	m.observations.EXPECT().List(gomock.Any(), testBuildID, afterIssue(issueIdentity(issue))).Return(nil, nil)
+	require.NoError(t, controller.processBuild(context.Background(), testQueue, testBuildID))
+}

stovepipe/entity/BUILD.bazel

@@ -7,9 +7,11 @@ go_library(
         "build_issue.go",
         "ingest.go",
         "issue_log.go",
+        "issue_state.go",
         "issue_tracking_state.go",
         "list.go",
         "open_issue.go",
+        "planned_issue_change.go",
         "project_fact_by_degree.go",
         "project_status.go",
         "queue.go",
@@ -36,7 +38,9 @@ go_test(
     srcs = [
         "build_test.go",
         "issue_log_test.go",
+        "issue_state_test.go",
         "issue_tracking_state_test.go",
+        "planned_issue_change_test.go",
         "project_fact_by_degree_test.go",
         "queue_state_test.go",
         "request_id_test.go",

stovepipe/entity/issue_state.go

@@ -0,0 +1,58 @@
+// 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
+
+import "fmt"
+
+// IssueStatus describes whether a validation issue is open or closed.
+type IssueStatus string
+
+const (
+	// IssueStatusUnknown is the unset status.
+	IssueStatusUnknown IssueStatus = ""
+	// IssueStatusOpen identifies an open validation issue.
+	IssueStatusOpen IssueStatus = "open"
+	// IssueStatusClosed identifies an issue that is no longer open.
+	IssueStatusClosed IssueStatus = "closed"
+)
+
+// IssueState is the latest state of a validation condition within a queue.
+// Closed records retain their identity and version across later openings.
+type IssueState struct {
+	// TaskID identifies the logical validation task and must be nonempty.
+	TaskID string `json:"task_id"`
+	// UnitID identifies the validation unit; empty for a task-wide issue.
+	UnitID string `json:"unit_id"`
+	// Status distinguishes open and closed issues.
+	Status IssueStatus `json:"status"`
+	// FirstSeenBuildID identifies the current or most recent opening.
+	FirstSeenBuildID string `json:"first_seen_build_id"`
+	// Version is the positive optimistic-lock version; it never resets on reopening.
+	Version int32 `json:"version"`
+}
+
+// Validate checks the identity, status, and version of an issue state.
+func (s IssueState) Validate() error {
+	if s.TaskID == "" || s.FirstSeenBuildID == "" {
+		return fmt.Errorf("issue state requires task and first-seen build IDs")
+	}
+	if s.Status != IssueStatusOpen && s.Status != IssueStatusClosed {
+		return fmt.Errorf("invalid issue status %q", s.Status)
+	}
+	if s.Version <= 0 {
+		return fmt.Errorf("issue state version must be positive")
+	}
+	return nil
+}

stovepipe/entity/issue_state_test.go

@@ -0,0 +1,46 @@
+// 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
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestIssueStateValidate(t *testing.T) {
+	valid := IssueState{TaskID: "task", FirstSeenBuildID: "build", Status: IssueStatusOpen, Version: 1}
+	for _, status := range []IssueStatus{IssueStatusOpen, IssueStatusClosed} {
+		value := valid
+		value.Status = status
+		require.NoError(t, value.Validate())
+	}
+	for _, field := range []string{"task", "build", "status", "version"} {
+		t.Run(field, func(t *testing.T) {
+			value := valid
+			switch field {
+			case "task":
+				value.TaskID = ""
+			case "build":
+				value.FirstSeenBuildID = ""
+			case "status":
+				value.Status = IssueStatusUnknown
+			case "version":
+				value.Version = 0
+			}
+			require.Error(t, value.Validate())
+		})
+	}
+}

stovepipe/entity/issue_tracking_state.go

@@ -20,7 +20,7 @@ import (
 	"github.com/uber/submitqueue/platform/base/id"
 )
 
-// IssueTrackingPhase describes whether one source build's issue updates are pending or complete.
+// IssueTrackingPhase describes whether one source build's issue updates are underway or complete.
 type IssueTrackingPhase string
 
 const (
@@ -30,7 +30,7 @@ const (
 	IssueTrackingPhaseRefreshing IssueTrackingPhase = "refreshing"
 	// IssueTrackingPhaseRefreshed means a refresh pass completed; later refresh notifications remain eligible.
 	IssueTrackingPhaseRefreshed IssueTrackingPhase = "refreshed"
-	// IssueTrackingPhaseFinalizing means final opening and closing updates are pending.
+	// IssueTrackingPhaseFinalizing means final opening and closing updates are underway.
 	IssueTrackingPhaseFinalizing IssueTrackingPhase = "finalizing"
 	// IssueTrackingPhaseFinalized means final reconciliation completed; later refresh notifications are ignored.
 	IssueTrackingPhaseFinalized IssueTrackingPhase = "finalized"
@@ -40,25 +40,25 @@ const (
 type IssueTrackingState struct {
 	// RequestID is the newest accepted source request, compared by ingest order.
 	RequestID string `json:"request_id"`
-	// BuildID identifies the source build whose pass is pending or most recently completed.
+	// BuildID identifies the source build whose pass is underway or most recently completed.
 	BuildID string `json:"build_id"`
 	// NotificationID identifies the message that owns this reconciliation pass.
 	NotificationID string `json:"notification_id"`
-	// AfterTaskID is the exclusive task cursor after fully processed issue keys.
-	// Empty with an empty AfterUnitID starts a pass.
-	AfterTaskID string `json:"after_task_id"`
-	// AfterUnitID is the unit component of the exclusive cursor; empty is task-wide.
-	AfterUnitID string `json:"after_unit_id"`
-	// Phase distinguishes pending work from completed refresh or final passes.
+	// CursorTaskID identifies the current task when a change is planned, or the last completed task.
+	// Empty with an empty CursorUnitID starts a pass.
+	CursorTaskID string `json:"cursor_task_id"`
+	// CursorUnitID identifies the unit at the cursor; empty is task-wide.
+	CursorUnitID string `json:"cursor_unit_id"`
+	// PlannedChange is the intended change for the issue at the cursor; zero means none.
+	PlannedChange PlannedIssueChange `json:"planned_change"`
+	// Phase distinguishes running passes from completed refresh or final passes.
 	Phase IssueTrackingPhase `json:"phase"`
 	// Version is the positive optimistic-lock version of this state.
 	Version int32 `json:"version"`
 }
 
-// IsPending reports whether the phase represents unfinished reconciliation.
-func (s IssueTrackingState) IsPending() bool {
-	return s.Phase == IssueTrackingPhaseRefreshing || s.Phase == IssueTrackingPhaseFinalizing
-}
+// HasPlannedChange reports whether a planned issue transition is present.
+func (s IssueTrackingState) HasPlannedChange() bool { return s.PlannedChange != (PlannedIssueChange{}) }
 
 // Validate verifies the identity, phase, and version of a persisted state.
 func (s IssueTrackingState) Validate() error {
@@ -71,16 +71,35 @@ func (s IssueTrackingState) Validate() error {
 	if s.NotificationID == "" {
 		return fmt.Errorf("issue tracking notification ID must not be empty")
 	}
-	if s.AfterTaskID == "" && s.AfterUnitID != "" {
+	if s.CursorTaskID == "" && s.CursorUnitID != "" {
 		return fmt.Errorf("issue tracking cursor unit requires a task")
 	}
 	if s.Version <= 0 {
 		return fmt.Errorf("issue tracking version must be positive")
 	}
 	switch s.Phase {
-	case IssueTrackingPhaseRefreshing, IssueTrackingPhaseRefreshed, IssueTrackingPhaseFinalizing, IssueTrackingPhaseFinalized:
-		return nil
+	case IssueTrackingPhaseRefreshing, IssueTrackingPhaseFinalizing:
+	case IssueTrackingPhaseRefreshed, IssueTrackingPhaseFinalized:
+		if s.HasPlannedChange() {
+			return fmt.Errorf("completed issue tracking cannot retain a planned change")
+		}
 	default:
 		return fmt.Errorf("invalid issue tracking phase %q", s.Phase)
 	}
+	if err := s.PlannedChange.Validate(); err != nil {
+		return err
+	}
+	if s.HasPlannedChange() {
+		if s.CursorTaskID == "" {
+			return fmt.Errorf("planned issue change requires a task at the cursor")
+		}
+		change := s.PlannedChange
+		if change.Status == IssueStatusOpen && change.FirstSeenBuildID != s.BuildID {
+			return fmt.Errorf("planned opening must identify its source build")
+		}
+		if change.Status == IssueStatusClosed && s.Phase != IssueTrackingPhaseFinalizing {
+			return fmt.Errorf("only final issue tracking can close an issue")
+		}
+	}
+	return nil
 }