diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index 80405d335..dd522ef07 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/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", diff --git a/stovepipe/entity/issue_tracking_state.go b/stovepipe/entity/issue_tracking_state.go index 789564f0f..558c4ebac 100644 --- a/stovepipe/entity/issue_tracking_state.go +++ b/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 } diff --git a/stovepipe/entity/issue_tracking_state_test.go b/stovepipe/entity/issue_tracking_state_test.go index d0287d57b..e0027e5e3 100644 --- a/stovepipe/entity/issue_tracking_state_test.go +++ b/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" }}, diff --git a/stovepipe/entity/planned_issue_change.go b/stovepipe/entity/planned_issue_change.go new file mode 100644 index 000000000..26a2f71cb --- /dev/null +++ b/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 +} diff --git a/stovepipe/entity/planned_issue_change_test.go b/stovepipe/entity/planned_issue_change_test.go new file mode 100644 index 000000000..75eb25083 --- /dev/null +++ b/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()) +} diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index c03c057b9..971f4f2d8 100644 --- a/stovepipe/extension/storage/README.md +++ b/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. diff --git a/stovepipe/extension/storage/mysql/issue_tracking_store.go b/stovepipe/extension/storage/mysql/issue_tracking_store.go index fd3ec329e..d8b1a3429 100644 --- a/stovepipe/extension/storage/mysql/issue_tracking_store.go +++ b/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) diff --git a/stovepipe/extension/storage/mysql/issue_tracking_store_test.go b/stovepipe/extension/storage/mysql/issue_tracking_store_test.go index 3d7e03825..3ef53e382 100644 --- a/stovepipe/extension/storage/mysql/issue_tracking_store_test.go +++ b/stovepipe/extension/storage/mysql/issue_tracking_store_test.go @@ -31,7 +31,7 @@ import ( ) func trackingFixture() entity.IssueTrackingState { - return entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1} + return entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", CursorTaskID: "task", CursorUnitID: "unit", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1, PlannedChange: entity.PlannedIssueChange{Status: entity.IssueStatusOpen, FirstSeenBuildID: "build/10", TimestampMs: 100}} } func TestIssueTrackingStoreGet(t *testing.T) { @@ -43,7 +43,7 @@ func TestIssueTrackingStoreGet(t *testing.T) { state := trackingFixture() store := NewIssueTrackingStore(db, testMetrics(), "queue") - query := mock.ExpectQuery(regexp.QuoteMeta(`SELECT request_id, build_id, notification_id, after_task_id, after_unit_id, phase, version FROM issue_tracking WHERE queue = ?`)).WithArgs("queue") + query := mock.ExpectQuery(regexp.QuoteMeta(`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 = ?`)).WithArgs("queue") switch kind { case "missing": query.WillReturnError(sql.ErrNoRows) @@ -54,7 +54,7 @@ func TestIssueTrackingStoreGet(t *testing.T) { if kind == "scan error" { request = nil } - query.WillReturnRows(sqlmock.NewRows([]string{"request_id", "build_id", "notification_id", "after_task_id", "after_unit_id", "phase", "version"}).AddRow(request, state.BuildID, state.NotificationID, state.AfterTaskID, state.AfterUnitID, state.Phase, state.Version)) + query.WillReturnRows(sqlmock.NewRows([]string{"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"}).AddRow(request, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.PlannedChange.Status, state.PlannedChange.FirstSeenBuildID, state.PlannedChange.ExpectedVersion, state.PlannedChange.TimestampMs, state.Phase, state.Version)) } got, err := store.Get(context.Background()) @@ -94,8 +94,8 @@ func TestIssueTrackingStoreCreate(t *testing.T) { state := trackingFixture() store := NewIssueTrackingStore(db, testMetrics(), "queue") - query := mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO issue_tracking (queue, request_id, build_id, notification_id, after_task_id, after_unit_id, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`)). - WithArgs("queue", state.RequestID, state.BuildID, state.NotificationID, state.AfterTaskID, state.AfterUnitID, state.Phase, state.Version) + query := mock.ExpectExec(regexp.QuoteMeta(`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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)). + WithArgs("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 tt.dbErr != nil { query.WillReturnError(tt.dbErr) } else { @@ -133,8 +133,8 @@ func TestIssueTrackingStoreUpdate(t *testing.T) { state := trackingFixture() store := NewIssueTrackingStore(db, testMetrics(), "queue") - query := mock.ExpectExec(regexp.QuoteMeta(`UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?, after_task_id = ?, after_unit_id = ?, phase = ?, version = ? WHERE queue = ? AND version = ?`)). - WithArgs(state.RequestID, state.BuildID, state.NotificationID, state.AfterTaskID, state.AfterUnitID, state.Phase, int32(7), "queue", int32(1)) + query := mock.ExpectExec(regexp.QuoteMeta(`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 = ?`)). + WithArgs(state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.PlannedChange.Status, state.PlannedChange.FirstSeenBuildID, state.PlannedChange.ExpectedVersion, state.PlannedChange.TimestampMs, state.Phase, int32(7), "queue", int32(1)) switch { case tt.dbErr != nil: query.WillReturnError(tt.dbErr) @@ -165,10 +165,17 @@ func TestIssueTrackingStoreRejectsInvalidState(t *testing.T) { {"invalid request", func(s *entity.IssueTrackingState, _ *string) { s.RequestID = "invalid" }}, {"empty notification", func(s *entity.IssueTrackingState, _ *string) { s.NotificationID = "" }}, {"oversized notification", func(s *entity.IssueTrackingState, _ *string) { s.NotificationID = strings.Repeat("n", 1021) }}, - {"oversized task cursor", func(s *entity.IssueTrackingState, _ *string) { s.AfterTaskID = strings.Repeat("t", 256) }}, + {"oversized task cursor", func(s *entity.IssueTrackingState, _ *string) { s.CursorTaskID = strings.Repeat("t", 256) }}, {"oversized unit cursor", func(s *entity.IssueTrackingState, _ *string) { - s.AfterTaskID = "task" - s.AfterUnitID = strings.Repeat("u", 1025) + s.CursorTaskID = "task" + s.CursorUnitID = strings.Repeat("u", 1025) + }}, + + {"oversized planned opening", func(s *entity.IssueTrackingState, _ *string) { + s.Phase = entity.IssueTrackingPhaseFinalizing + s.PlannedChange.Status = entity.IssueStatusClosed + s.PlannedChange.ExpectedVersion = 1 + s.PlannedChange.FirstSeenBuildID = strings.Repeat("b", 1021) }}, {"empty build", func(s *entity.IssueTrackingState, _ *string) { s.BuildID = "" }}, {"oversized build", func(s *entity.IssueTrackingState, _ *string) { s.BuildID = strings.Repeat("b", 1021) }}, diff --git a/stovepipe/extension/storage/mysql/schema/issue_tracking.sql b/stovepipe/extension/storage/mysql/schema/issue_tracking.sql index 565630ddc..38caaf0fe 100644 --- a/stovepipe/extension/storage/mysql/schema/issue_tracking.sql +++ b/stovepipe/extension/storage/mysql/schema/issue_tracking.sql @@ -1,12 +1,19 @@ --- Persistent source-order boundary; pending passes are resumed before admitting another source. +-- Persistent source-order boundary; unfinished passes are resumed before admitting another source. CREATE TABLE IF NOT EXISTS issue_tracking ( - queue VARBINARY(255) NOT NULL, - request_id VARBINARY(255) NOT NULL, - build_id VARBINARY(1020) NOT NULL, - notification_id VARBINARY(1020) NOT NULL, - after_task_id VARBINARY(255) NOT NULL, - after_unit_id VARBINARY(1024) NOT NULL, - phase VARBINARY(32) NOT NULL, - version INT NOT NULL, + queue VARBINARY(255) NOT NULL, + request_id VARBINARY(255) NOT NULL, + build_id VARBINARY(1020) NOT NULL, + notification_id VARBINARY(1020) NOT NULL, + + cursor_task_id VARBINARY(255) NOT NULL, + cursor_unit_id VARBINARY(1024) NOT NULL, + + planned_status VARBINARY(32) NOT NULL DEFAULT '', + planned_first_seen_build_id VARBINARY(1020) NOT NULL DEFAULT '', + planned_expected_version INT NOT NULL DEFAULT 0, + planned_timestamp_ms BIGINT NOT NULL DEFAULT 0, + + phase VARBINARY(32) NOT NULL, + version INT NOT NULL, PRIMARY KEY (queue) ) ENGINE=InnoDB; diff --git a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go index ce9e90e39..11fb862c2 100644 --- a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go @@ -39,7 +39,7 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F require.NoError(t, err) require.Equal(t, state, got) - state.AfterTaskID, state.AfterUnitID = "task", "unit" + state.CursorTaskID, state.CursorUnitID = "task", "unit" state.Phase = entity.IssueTrackingPhaseRefreshed require.NoError(t, store.Update(ctx, state, 1, 2)) state.Version = 2 @@ -61,6 +61,26 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F require.NoError(t, err) require.Empty(t, opens) + t.Run("clearing a planned change retains its cursor", func(t *testing.T) { + bound, err := factory.For(storage.Config{QueueName: "tracking/planned"}) + require.NoError(t, err) + store := bound.GetIssueTrackingStore() + progress := entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notice", CursorTaskID: "task", CursorUnitID: "unit", Phase: entity.IssueTrackingPhaseFinalizing, Version: 1, + PlannedChange: entity.PlannedIssueChange{Status: entity.IssueStatusClosed, FirstSeenBuildID: "older", ExpectedVersion: 7, TimestampMs: 1234}} + require.NoError(t, store.Create(ctx, progress)) + retained, err := store.Get(ctx) + require.NoError(t, err) + require.Equal(t, progress, retained) + next := progress + next.PlannedChange = entity.PlannedIssueChange{} + require.NoError(t, store.Update(ctx, next, 1, 2)) + require.ErrorIs(t, store.Update(ctx, progress, 1, 2), storage.ErrVersionMismatch) + retained, err = store.Get(ctx) + require.NoError(t, err) + require.False(t, retained.HasPlannedChange()) + require.Equal(t, "task", retained.CursorTaskID) + require.Equal(t, "unit", retained.CursorUnitID) + }) t.Run("queue isolation", func(t *testing.T) { other, err := factory.For(storage.Config{QueueName: "Tracking/main"}) require.NoError(t, err) @@ -105,7 +125,7 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F bound, err := factory.For(storage.Config{QueueName: strings.Repeat("q", 255)}) require.NoError(t, err) store := bound.GetIssueTrackingStore() - value := entity.IssueTrackingState{RequestID: "9223372036854775807", BuildID: strings.Repeat("😀", 255), NotificationID: strings.Repeat("😀", 255), AfterTaskID: strings.Repeat("t", 255), AfterUnitID: strings.Repeat("u", 1024), Phase: entity.IssueTrackingPhaseRefreshing, Version: 1} + value := entity.IssueTrackingState{RequestID: "9223372036854775807", BuildID: strings.Repeat("😀", 255), NotificationID: strings.Repeat("😀", 255), CursorTaskID: strings.Repeat("t", 255), CursorUnitID: strings.Repeat("u", 1024), Phase: entity.IssueTrackingPhaseRefreshing, Version: 1} require.NoError(t, store.Create(ctx, value)) got, err := store.Get(ctx) require.NoError(t, err)