feat(stovepipe): Save pending issue changes

Author: prathshenoy

stovepipe/entity/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
         "issue_tracking_state.go",
         "list.go",
         "open_issue.go",
+        "pending_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",
+        "pending_issue_change_test.go",
         "project_fact_by_degree_test.go",
         "queue_state_test.go",
         "request_id_test.go",

stovepipe/entity/issue_tracking_state.go

@@ -49,6 +49,8 @@ type IssueTrackingState struct {
 	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"`
+	// PendingChange is the saved issue transition; its zero value means none.
+	PendingChange PendingIssueChange `json:"pending_change"`
 	// Phase distinguishes pending work from completed refresh or final passes.
 	Phase IssueTrackingPhase `json:"phase"`
 	// Version is the positive optimistic-lock version of this state.
@@ -60,6 +62,9 @@ func (s IssueTrackingState) IsPending() bool {
 	return s.Phase == IssueTrackingPhaseRefreshing || s.Phase == IssueTrackingPhaseFinalizing
 }
 
+// HasPendingChange reports whether a saved issue transition is present.
+func (s IssueTrackingState) HasPendingChange() bool { return s.PendingChange.TaskID != "" }
+
 // Validate verifies the identity, phase, and version of a persisted state.
 func (s IssueTrackingState) Validate() error {
 	if err := id.Validate(s.RequestID); err != nil {
@@ -77,6 +82,24 @@ func (s IssueTrackingState) Validate() error {
 	if s.Version <= 0 {
 		return fmt.Errorf("issue tracking version must be positive")
 	}
+	if err := s.PendingChange.Validate(); err != nil {
+		return err
+	}
+	if s.HasPendingChange() {
+		if !s.IsPending() {
+			return fmt.Errorf("completed issue tracking cannot retain a pending change")
+		}
+		change := s.PendingChange
+		if change.TaskID < s.AfterTaskID || (change.TaskID == s.AfterTaskID && change.UnitID <= s.AfterUnitID) {
+			return fmt.Errorf("pending issue change must follow the completed cursor")
+		}
+		if change.Status == IssueStatusOpen && change.FirstSeenBuildID != s.BuildID {
+			return fmt.Errorf("pending opening must identify its source build")
+		}
+		if change.Status == IssueStatusClosed && s.Phase != IssueTrackingPhaseFinalizing {
+			return fmt.Errorf("only final issue tracking can close an issue")
+		}
+	}
 	switch s.Phase {
 	case IssueTrackingPhaseRefreshing, IssueTrackingPhaseRefreshed, IssueTrackingPhaseFinalizing, IssueTrackingPhaseFinalized:
 		return nil

stovepipe/entity/pending_issue_change.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"
+	"math"
+)
+
+// PendingIssueChange is one saved issue transition with its original CAS guard.
+// The zero value means no change is pending.
+type PendingIssueChange struct {
+	// TaskID identifies the issue's validation task.
+	TaskID string `json:"task_id"`
+	// UnitID identifies the unit; empty is task-wide.
+	UnitID string `json:"unit_id"`
+	// 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 PendingIssueChange) Validate() error {
+	if c == (PendingIssueChange{}) {
+		return nil
+	}
+	if c.ExpectedVersion < 0 || c.ExpectedVersion == math.MaxInt32 {
+		return fmt.Errorf("pending issue version must permit a positive successor")
+	}
+	target := IssueState{TaskID: c.TaskID, UnitID: c.UnitID, Status: c.Status, FirstSeenBuildID: c.FirstSeenBuildID, Version: c.ExpectedVersion + 1}
+	if err := target.Validate(); err != nil {
+		return err
+	}
+	if c.Status == IssueStatusClosed && c.ExpectedVersion == 0 {
+		return fmt.Errorf("closing an issue requires an existing version")
+	}
+	if c.TimestampMs <= 0 {
+		return fmt.Errorf("pending issue timestamp must be positive")
+	}
+	return nil
+}

stovepipe/entity/pending_issue_change_test.go

@@ -0,0 +1,73 @@
+// 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 TestPendingIssueChangeValidate(t *testing.T) {
+	require.NoError(t, (PendingIssueChange{}).Validate())
+	opening := PendingIssueChange{TaskID: "task", 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(*PendingIssueChange)
+	}{
+		{"missing task", func(c *PendingIssueChange) { c.TaskID = "" }},
+		{"missing opening", func(c *PendingIssueChange) { c.FirstSeenBuildID = "" }},
+		{"unknown status", func(c *PendingIssueChange) { c.Status = IssueStatusUnknown }},
+		{"negative guard", func(c *PendingIssueChange) { c.ExpectedVersion = -1 }},
+		{"exhausted guard", func(c *PendingIssueChange) { c.ExpectedVersion = math.MaxInt32 }},
+		{"closing absent issue", func(c *PendingIssueChange) { c.Status = IssueStatusClosed }},
+		{"missing timestamp", func(c *PendingIssueChange) { c.TimestampMs = 0 }},
+	} {
+		t.Run(tt.name, func(t *testing.T) { value := opening; tt.change(&value); require.Error(t, value.Validate()) })
+	}
+}
+
+func TestIssueTrackingPendingChange(t *testing.T) {
+	state := IssueTrackingState{RequestID: "10", BuildID: "build", NotificationID: "notice", Phase: IssueTrackingPhaseRefreshing, Version: 1,
+		PendingChange: PendingIssueChange{TaskID: "task", UnitID: "unit", Status: IssueStatusOpen, FirstSeenBuildID: "build", TimestampMs: 100}}
+	require.True(t, state.HasPendingChange())
+	require.NoError(t, state.Validate())
+	for _, tt := range []struct {
+		name   string
+		change func(*IssueTrackingState)
+	}{
+		{"completed pass", func(s *IssueTrackingState) { s.Phase = IssueTrackingPhaseRefreshed }},
+		{"behind cursor", func(s *IssueTrackingState) { s.AfterTaskID = "z" }},
+		{"at cursor", func(s *IssueTrackingState) { s.AfterTaskID = "task"; s.AfterUnitID = "unit" }},
+		{"opening from another build", func(s *IssueTrackingState) { s.PendingChange.FirstSeenBuildID = "other" }},
+		{"refresh cannot close", func(s *IssueTrackingState) {
+			s.PendingChange.Status = IssueStatusClosed
+			s.PendingChange.ExpectedVersion = 1
+		}},
+	} {
+		t.Run(tt.name, func(t *testing.T) { value := state; tt.change(&value); require.Error(t, value.Validate()) })
+	}
+	state.Phase = IssueTrackingPhaseFinalizing
+	state.PendingChange.Status = IssueStatusClosed
+	state.PendingChange.ExpectedVersion = 7
+	state.PendingChange.FirstSeenBuildID = "older"
+	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, 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. 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 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.
+
+### Saved issue changes
+
+Progress can retain one `PendingIssueChange`: the task/unit, intended status and first-seen build, original expected issue version, and history timestamp. 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 pending change is saved 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 and advancing the cursor. Pending 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,
+            after_task_id, after_unit_id,
+            pending_task_id, pending_unit_id, pending_status, pending_first_seen_build_id,
+            pending_expected_version, pending_timestamp_ms, phase, version
+         FROM issue_tracking WHERE queue = ?`, s.queue).
+		Scan(
+			&state.RequestID, &state.BuildID, &state.NotificationID,
+			&state.AfterTaskID, &state.AfterUnitID,
+			&state.PendingChange.TaskID, &state.PendingChange.UnitID, &state.PendingChange.Status, &state.PendingChange.FirstSeenBuildID,
+			&state.PendingChange.ExpectedVersion, &state.PendingChange.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,
+            after_task_id, after_unit_id,
+            pending_task_id, pending_unit_id, pending_status, pending_first_seen_build_id,
+            pending_expected_version, pending_timestamp_ms, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		s.queue, state.RequestID, state.BuildID, state.NotificationID,
+		state.AfterTaskID, state.AfterUnitID,
+		state.PendingChange.TaskID, state.PendingChange.UnitID, state.PendingChange.Status, state.PendingChange.FirstSeenBuildID,
+		state.PendingChange.ExpectedVersion, state.PendingChange.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 = ?,
+            after_task_id = ?, after_unit_id = ?,
+            pending_task_id = ?, pending_unit_id = ?, pending_status = ?, pending_first_seen_build_id = ?,
+            pending_expected_version = ?, pending_timestamp_ms = ?, phase = ?, version = ?
+         WHERE queue = ? AND version = ?`,
+		state.RequestID, state.BuildID, state.NotificationID,
+		state.AfterTaskID, state.AfterUnitID,
+		state.PendingChange.TaskID, state.PendingChange.UnitID, state.PendingChange.Status, state.PendingChange.FirstSeenBuildID,
+		state.PendingChange.ExpectedVersion, state.PendingChange.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.AfterTaskID, state.AfterUnitID); err != nil {
+		return err
+	}
+	change := state.PendingChange
+	return validateBuildIssueKey(s.queue, change.FirstSeenBuildID, change.TaskID, change.UnitID)
 }
 
 var _ storage.IssueTrackingStore = (*issueTrackingStore)(nil)