feat(stovepipe): Retain versioned issue state

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,6 +7,7 @@ go_library(
         "build_issue.go",
         "ingest.go",
         "issue_log.go",
+        "issue_state.go",
         "issue_tracking_state.go",
         "list.go",
         "open_issue.go",
@@ -36,6 +37,7 @@ go_test(
     srcs = [
         "build_test.go",
         "issue_log_test.go",
+        "issue_state_test.go",
         "issue_tracking_state_test.go",
         "project_fact_by_degree_test.go",
         "queue_state_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/open_issue.go

@@ -16,7 +16,7 @@ package entity
 
 // OpenIssue is a currently open validation issue within a queue.
 // Its task and unit identify the condition; the first-seen build identifies
-// this opening and is immutable until removal.
+// this opening.
 type OpenIssue struct {
 	// TaskID identifies the logical validation task and must be nonempty.
 	TaskID string `json:"task_id"`

stovepipe/extension/storage/BUILD.bazel

@@ -6,8 +6,8 @@ go_library(
         "build_issue_store.go",
         "build_store.go",
         "issue_log_store.go",
+        "issue_state_store.go",
         "issue_tracking_store.go",
-        "open_issue_store.go",
         "project_fact_by_degree_store.go",
         "queue_log_store.go",
         "queue_store.go",