feat(stovepipe): Save planned issue changes

stovepipe/entity/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
         "issue_tracking_state.go",
         "list.go",
         "open_issue.go",
+        "planned_issue_change.go",
         "project_fact_by_degree.go",
         "project_status.go",
         "queue.go",
@@ -39,6 +40,7 @@ go_test(
         "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_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
 }

stovepipe/entity/issue_tracking_state_test.go

@@ -25,7 +25,7 @@ func TestIssueTrackingState(t *testing.T) {
 		t.Run(string(phase), func(t *testing.T) {
 			state := IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", Phase: phase, Version: 1}
 			require.NoError(t, state.Validate())
-			require.Equal(t, phase == IssueTrackingPhaseRefreshing || phase == IssueTrackingPhaseFinalizing, state.IsPending())
+			require.False(t, state.HasPlannedChange())
 		})
 	}
 	for _, tt := range []struct {
@@ -36,7 +36,7 @@ func TestIssueTrackingState(t *testing.T) {
 		{"noncanonical request", func(s *IssueTrackingState) { s.RequestID = "010" }},
 		{"nonnumeric request", func(s *IssueTrackingState) { s.RequestID = "request/10" }},
 		{"missing notification", func(s *IssueTrackingState) { s.NotificationID = "" }},
-		{"unit cursor without task", func(s *IssueTrackingState) { s.AfterUnitID = "unit" }},
+		{"unit cursor without task", func(s *IssueTrackingState) { s.CursorUnitID = "unit" }},
 		{"missing build", func(s *IssueTrackingState) { s.BuildID = "" }},
 		{"unknown phase", func(s *IssueTrackingState) { s.Phase = IssueTrackingPhaseUnknown }},
 		{"unsupported phase", func(s *IssueTrackingState) { s.Phase = "other" }},

stovepipe/entity/planned_issue_change.go

@@ -0,0 +1,56 @@
+// 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"
+	"math"
+)
+
+// PlannedIssueChange describes an intended issue transition and its original version guard.
+// The zero value means no change is planned.
+type PlannedIssueChange struct {
+	// Status is the intended open or closed status.
+	Status IssueStatus `json:"status"`
+	// FirstSeenBuildID identifies the opening retained by the intended issue state.
+	FirstSeenBuildID string `json:"first_seen_build_id"`
+	// ExpectedVersion is the original issue version; zero requires absence.
+	ExpectedVersion int32 `json:"expected_version"`
+	// TimestampMs is the transition's retained history timestamp in Unix milliseconds.
+	TimestampMs int64 `json:"timestamp_ms"`
+}
+
+// Validate checks a complete change or the empty sentinel.
+func (c PlannedIssueChange) Validate() error {
+	if c == (PlannedIssueChange{}) {
+		return nil
+	}
+	if c.ExpectedVersion < 0 || c.ExpectedVersion == math.MaxInt32 {
+		return fmt.Errorf("planned issue version must permit a positive successor")
+	}
+	if c.Status != IssueStatusOpen && c.Status != IssueStatusClosed {
+		return fmt.Errorf("invalid planned issue status %q", c.Status)
+	}
+	if c.FirstSeenBuildID == "" {
+		return fmt.Errorf("planned issue change requires a first-seen build ID")
+	}
+	if c.Status == IssueStatusClosed && c.ExpectedVersion == 0 {
+		return fmt.Errorf("closing an issue requires an existing version")
+	}
+	if c.TimestampMs <= 0 {
+		return fmt.Errorf("planned issue timestamp must be positive")
+	}
+	return nil
+}

stovepipe/entity/planned_issue_change_test.go

@@ -0,0 +1,81 @@
+// 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 (
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestPlannedIssueChangeValidate(t *testing.T) {
+	require.NoError(t, (PlannedIssueChange{}).Validate())
+	opening := PlannedIssueChange{Status: IssueStatusOpen, FirstSeenBuildID: "build", TimestampMs: 100}
+	require.NoError(t, opening.Validate())
+	closing := opening
+	closing.Status = IssueStatusClosed
+	closing.ExpectedVersion = 7
+	require.NoError(t, closing.Validate())
+	for _, tt := range []struct {
+		name   string
+		change func(*PlannedIssueChange)
+	}{
+		{"missing opening", func(c *PlannedIssueChange) { c.FirstSeenBuildID = "" }},
+		{"unknown status", func(c *PlannedIssueChange) { c.Status = IssueStatusUnknown }},
+		{"unsupported status", func(c *PlannedIssueChange) { c.Status = "other" }},
+		{"negative guard", func(c *PlannedIssueChange) { c.ExpectedVersion = -1 }},
+		{"exhausted guard", func(c *PlannedIssueChange) { c.ExpectedVersion = math.MaxInt32 }},
+		{"closing absent issue", func(c *PlannedIssueChange) { c.Status = IssueStatusClosed }},
+		{"missing timestamp", func(c *PlannedIssueChange) { c.TimestampMs = 0 }},
+	} {
+		t.Run(tt.name, func(t *testing.T) { value := opening; tt.change(&value); require.Error(t, value.Validate()) })
+	}
+}
+
+func TestIssueTrackingPlannedChange(t *testing.T) {
+	state := IssueTrackingState{RequestID: "10", BuildID: "build", NotificationID: "notice", CursorTaskID: "task", CursorUnitID: "unit", Phase: IssueTrackingPhaseRefreshing, Version: 1,
+		PlannedChange: PlannedIssueChange{Status: IssueStatusOpen, FirstSeenBuildID: "build", TimestampMs: 100}}
+	require.True(t, state.HasPlannedChange())
+	require.NoError(t, state.Validate())
+	for _, tt := range []struct {
+		name   string
+		change func(*IssueTrackingState)
+	}{
+		{"completed pass", func(s *IssueTrackingState) { s.Phase = IssueTrackingPhaseRefreshed }},
+		{"finalized pass", func(s *IssueTrackingState) { s.Phase = IssueTrackingPhaseFinalized }},
+		{"missing cursor", func(s *IssueTrackingState) { s.CursorTaskID, s.CursorUnitID = "", "" }},
+		{"opening from another build", func(s *IssueTrackingState) { s.PlannedChange.FirstSeenBuildID = "other" }},
+		{"refresh cannot close", func(s *IssueTrackingState) {
+			s.PlannedChange.Status = IssueStatusClosed
+			s.PlannedChange.ExpectedVersion = 1
+		}},
+	} {
+		t.Run(tt.name, func(t *testing.T) { value := state; tt.change(&value); require.Error(t, value.Validate()) })
+	}
+	state.Phase = IssueTrackingPhaseFinalizing
+	state.PlannedChange.Status = IssueStatusClosed
+	state.PlannedChange.ExpectedVersion = 7
+	state.PlannedChange.FirstSeenBuildID = "older"
+	require.NoError(t, state.Validate())
+	state.CursorUnitID = ""
+	require.NoError(t, state.Validate())
+
+	state.PlannedChange = PlannedIssueChange{}
+	require.False(t, state.HasPlannedChange())
+	require.Equal(t, "task", state.CursorTaskID)
+	state.Phase = IssueTrackingPhaseFinalized
+	require.NoError(t, state.Validate())
+}

stovepipe/extension/storage/README.md

@@ -62,8 +62,14 @@ History persistence does not alter issue state or validate transitions against i
 
 ## Issue reconciliation progress
 
-`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, whether refresh or final writes are pending or complete, the owning notification, an exclusive task/unit cursor, and its optimistic version. The zero cursor starts a pass. A cursor checkpoint follows the writes for every processed key; notification identity lets the controller distinguish its own continuation from helping another pass. Keeping it after the last open row is removed preserves the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction.
+`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, the refresh or finalization phase, the owning notification, one task/unit cursor, and its optimistic version. The zero cursor starts a pass. With a planned change, the cursor identifies the issue whose writes must finish; without one, it identifies the last completed key. Notification identity lets the controller distinguish its own continuation from helping another pass. The record remains after all issues close to preserve the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction.
 
-The store does not decide source ordering, apply issue transitions, or advance versions. Its caller supplies the guard version and replacement version. A later controller will own acceptance, resumption of pending writes, and completion; no runtime writer is enabled here.
+The store does not decide source ordering, apply issue transitions, or advance versions. Its caller supplies the guard version and replacement version. A later controller will own acceptance, resumption of planned changes, and completion; no runtime writer is enabled here.
 
 The issue-tracking table must include the notification and cursor columns before updated readers or writers are deployed. No runtime writer is enabled by the storage layer. Notification IDs allow up to 1020 bytes, matching the queue message ID byte capacity; cursor limits match the issue stores.
+
+### Planned issue changes
+
+Progress can retain one `PlannedIssueChange`: the intended status and first-seen build, original expected issue version, and history timestamp for the issue at the cursor. The task/unit key exists only in the cursor, not in the change. Expected version zero requires create-if-absent; a positive value guards an update of an existing record. The expected version is never replaced with a newer value when a retry resumes old instructions.
+
+The cursor and change are saved together with progress CAS before any issue write. A replacement worker must finish that same issue write and its idempotent history event before clearing the change. Clearing retains the cursor, so the next scan continues after the completed issue. Planned changes are bounded fields, not a serialized batch. Closing retains the issue record; neither issue versions nor the queue's source boundary may be reset while old deliveries could resume.

stovepipe/extension/storage/mysql/issue_tracking_store.go

@@ -41,8 +41,17 @@ func (s *issueTrackingStore) Get(ctx context.Context) (state entity.IssueTrackin
 	op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets)
 	defer func() { op.Complete(retErr) }()
 	err := s.db.QueryRowContext(ctx,
-		`SELECT request_id, build_id, notification_id, after_task_id, after_unit_id, phase, version FROM issue_tracking WHERE queue = ?`, s.queue).
-		Scan(&state.RequestID, &state.BuildID, &state.NotificationID, &state.AfterTaskID, &state.AfterUnitID, &state.Phase, &state.Version)
+		`SELECT request_id, build_id, notification_id,
+            cursor_task_id, cursor_unit_id,
+            planned_status, planned_first_seen_build_id,
+            planned_expected_version, planned_timestamp_ms, phase, version
+         FROM issue_tracking WHERE queue = ?`, s.queue).
+		Scan(
+			&state.RequestID, &state.BuildID, &state.NotificationID,
+			&state.CursorTaskID, &state.CursorUnitID,
+			&state.PlannedChange.Status, &state.PlannedChange.FirstSeenBuildID,
+			&state.PlannedChange.ExpectedVersion, &state.PlannedChange.TimestampMs, &state.Phase, &state.Version,
+		)
 	if errors.Is(err, sql.ErrNoRows) {
 		return entity.IssueTrackingState{}, storage.WrapNotFound(err)
 	}
@@ -59,8 +68,15 @@ func (s *issueTrackingStore) Create(ctx context.Context, state entity.IssueTrack
 		return err
 	}
 	_, err := s.db.ExecContext(ctx,
-		`INSERT INTO issue_tracking (queue, request_id, build_id, notification_id, after_task_id, after_unit_id, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
-		s.queue, state.RequestID, state.BuildID, state.NotificationID, state.AfterTaskID, state.AfterUnitID, state.Phase, state.Version)
+		`INSERT INTO issue_tracking (queue, request_id, build_id, notification_id,
+            cursor_task_id, cursor_unit_id,
+            planned_status, planned_first_seen_build_id,
+            planned_expected_version, planned_timestamp_ms, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		s.queue, state.RequestID, state.BuildID, state.NotificationID,
+		state.CursorTaskID, state.CursorUnitID,
+		state.PlannedChange.Status, state.PlannedChange.FirstSeenBuildID,
+		state.PlannedChange.ExpectedVersion, state.PlannedChange.TimestampMs, state.Phase, state.Version,
+	)
 	if isDuplicateEntry(err) {
 		return storage.ErrAlreadyExists
 	}
@@ -82,8 +98,17 @@ func (s *issueTrackingStore) Update(ctx context.Context, state entity.IssueTrack
 		return fmt.Errorf("issue tracking guard version must be positive")
 	}
 	result, err := s.db.ExecContext(ctx,
-		`UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?, after_task_id = ?, after_unit_id = ?, phase = ?, version = ? WHERE queue = ? AND version = ?`,
-		state.RequestID, state.BuildID, state.NotificationID, state.AfterTaskID, state.AfterUnitID, state.Phase, newVersion, s.queue, oldVersion)
+		`UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?,
+            cursor_task_id = ?, cursor_unit_id = ?,
+            planned_status = ?, planned_first_seen_build_id = ?,
+            planned_expected_version = ?, planned_timestamp_ms = ?, phase = ?, version = ?
+         WHERE queue = ? AND version = ?`,
+		state.RequestID, state.BuildID, state.NotificationID,
+		state.CursorTaskID, state.CursorUnitID,
+		state.PlannedChange.Status, state.PlannedChange.FirstSeenBuildID,
+		state.PlannedChange.ExpectedVersion, state.PlannedChange.TimestampMs, state.Phase, newVersion,
+		s.queue, oldVersion,
+	)
 	if err != nil {
 		return fmt.Errorf("update issue tracking state: %w", err)
 	}
@@ -105,7 +130,11 @@ func (s *issueTrackingStore) validateState(state entity.IssueTrackingState) erro
 	if len(state.NotificationID) > 1020 {
 		return fmt.Errorf("issue tracking notification ID exceeds 1020-byte limit")
 	}
-	return validateBuildIssueKey(s.queue, state.BuildID, state.AfterTaskID, state.AfterUnitID)
+	if err := validateBuildIssueKey(s.queue, state.BuildID, state.CursorTaskID, state.CursorUnitID); err != nil {
+		return err
+	}
+	change := state.PlannedChange
+	return validateBuildIssueKey(s.queue, change.FirstSeenBuildID, state.CursorTaskID, state.CursorUnitID)
 }
 
 var _ storage.IssueTrackingStore = (*issueTrackingStore)(nil)