diff --git a/stovepipe/controller/fix/fix.go b/stovepipe/controller/fix/fix.go index 51b331828..55543d220 100644 --- a/stovepipe/controller/fix/fix.go +++ b/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 } } diff --git a/stovepipe/controller/fix/fix_test.go b/stovepipe/controller/fix/fix_test.go index a1c575245..25f3af813 100644 --- a/stovepipe/controller/fix/fix_test.go +++ b/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)) +} diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index 444eb0a73..dd522ef07 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/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", diff --git a/stovepipe/entity/issue_state.go b/stovepipe/entity/issue_state.go new file mode 100644 index 000000000..aa5363857 --- /dev/null +++ b/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 +} diff --git a/stovepipe/entity/issue_state_test.go b/stovepipe/entity/issue_state_test.go new file mode 100644 index 000000000..806445283 --- /dev/null +++ b/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()) + }) + } +} 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/open_issue.go b/stovepipe/entity/open_issue.go index 11e8f8c89..c1c0ed269 100644 --- a/stovepipe/entity/open_issue.go +++ b/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"` 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/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index c924bd1c5..ebdcb1015 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/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", diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index bb1d47a6a..971f4f2d8 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -1,6 +1,6 @@ # Storage -Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`, `BuildIssueStore`, `OpenIssueStore`, `IssueLogStore`, `IssueTrackingStore`). Implementations live under `extension/storage//`. +Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`, `BuildIssueStore`, `IssueStateStore`, `IssueLogStore`, `IssueTrackingStore`). Implementations live under `extension/storage//`. The aggregate is resolved per queue through a factory keyed by queue name, mirroring the extension contract and the [submitqueue storage seam](../../../submitqueue/extension/storage/README.md#queue-scoped-resolution): a resolved instance is bound to its queue, entity arguments whose queue disagrees with the binding are rejected, and reads never surface another queue's records. Stovepipe has no cross-queue read paths, so every store — including `QueueStore`, whose row key is the queue name itself — lives inside the queue-scoped aggregate; there is no global remainder. @@ -8,7 +8,7 @@ This is a separate contract from `submitqueue/extension/storage` — same shape ## Optimistic locking contract -Entities that support concurrent mutation (`Request`, `Build`) carry an `int32 Version` field. `Update` methods take both `oldVersion` (the where-clause guard) and `newVersion` (the value to write) — the store performs a pure conditional write and never computes `oldVersion + 1` itself. Version arithmetic is owned by the controller: it computes `newVersion`, calls `Update`, and only assigns `entity.Version = newVersion` after the call succeeds. See [AGENTS.md](../../../AGENTS.md) and the [submitqueue storage README](../../../submitqueue/extension/storage/README.md#optimistic-locking-contract) for the full rationale and the caller pattern — the convention is identical here. +Entities that support concurrent mutation (`Request`, `Build`, `IssueState`, `IssueTrackingState`) carry an `int32 Version` field. `Update` methods take both `oldVersion` (the where-clause guard) and `newVersion` (the value to write) — the store performs a pure conditional write and never computes `oldVersion + 1` itself. Version arithmetic is owned by the controller: it computes `newVersion`, calls `Update`, and only assigns `entity.Version = newVersion` after the call succeeds. See [AGENTS.md](../../../AGENTS.md) and the [submitqueue storage README](../../../submitqueue/extension/storage/README.md#optimistic-locking-contract) for the full rationale and the caller pattern — the convention is identical here. ## Read-after-write consistency @@ -40,34 +40,36 @@ The MySQL primary key contains the queue, build ID, task ID, and unit ID directl Existing MySQL deployments must add `fix_id VARCHAR(255) NOT NULL DEFAULT ''` and `version INT NOT NULL DEFAULT 0` to `build_issue` before deploying the new readers and writers. The schema's `CREATE TABLE IF NOT EXISTS` does not alter existing tables. -## Currently open issues +## Issue state -`OpenIssueStore` contains only currently open issues, keyed by task and unit within the bound queue. Removing an open row never removes durable per-build observations. Task IDs must be nonempty; an empty unit identifies a task-wide issue. There is no open/resolved flag; a resolved issue has no row here. +`IssueStateStore` retains the latest open or closed state for each task/unit within its queue. The primary key remains `(queue, task, unit)`; status is a value, not a key component. A closed record retains its first-seen build and positive CAS version. Reopening updates the same record with a new first-seen build and a higher version. Records are never deleted, so a stale create cannot reuse an empty key after closure. -An open row is immutable. Its first-seen build identifies the current opening, not its latest observation. Repeated observations remain in the per-build store. Reopening after removal uses a different first-seen build ID; no last-observed build, version counter, or update operation is stored here. +Create is conditional on absence, including closed records. Update checks the supplied old version and writes the caller's replacement version. Controllers compute versions and update their in-memory values only after success. The store does not decide which source build is allowed to change an issue; that decision belongs to the issue-tracking controller. -Listing scans the bound queue's primary-key prefix in ascending bytewise task/unit order, returning at most the requested positive limit. Continue after the last returned task/unit key; the zero cursor starts the range. A cursor with a unit but no task is invalid. Empty units remain task-wide issues and can be used as continuation boundaries. No secondary index or schema change is required. +Listing reads a bounded primary-key range in bytewise task/unit order and includes both statuses. Callers filter closed records in memory and advance after the last scanned key, including closed records. A page containing only closed records is not the end of the range. No status index or server-side filtering is required. Task IDs must be nonempty; empty units identify task-wide issues. Identifier byte limits match the build-issue store. -Pages are not a snapshot across calls. Concurrent changes can affect which rows appear, and the cursor row need not still exist. Readers must not treat listing as an atomic replacement or resolution decision; guarded deletion still protects a later opening. - -Deletion matches the task, unit, and first-seen build within the bound queue. It is safe to repeat: an absent opening is a no-op, and a stale deletion never removes a different opening of the same condition. Database failures are returned for retry. - -MySQL identifiers use the same binary byte limits as per-build observations. Build IDs are opaque context, not ordering tokens. The store does not choose resolution, write lifecycle history, or reconcile analysis responses. Lifecycle occurrences are retained separately by `IssueLogStore`. Source-order reconciliation belongs to later controllers, not this membership store. +The MySQL backend uses the new `issue_state` table. Create it before deploying the updated readers. The prior `open_issue` table is not read or automatically copied; no in-repository runtime writer was enabled for it. Deployments with externally populated rows need an explicit backfill before switching readers. The repair executor still receives an `OpenIssue`, which describes an open occurrence rather than the storage record. ## Issue lifecycle history -`IssueLogStore` retains immutable opened and resolved transitions independently of current open-issue membership and per-build observations. Each transition is keyed by task, unit, build, and event within the bound queue. An empty unit identifies a task-wide issue. History remains after an open row is removed or recreated; transitions do not carry a separate reference to the opening they belong to. +`IssueLogStore` retains immutable opened and resolved transitions independently of current issue state and per-build observations. Each transition is keyed by task, unit, build, and event within the bound queue. An empty unit identifies a task-wide issue. History remains when an issue closes or reopens; transitions do not carry a separate reference to the opening they belong to. The build ID identifies the build where the issue was first seen or confirmed resolved. The application supplies a positive Unix-millisecond timestamp for when the transition was first recorded, not when the build finished. A duplicate identity returns `ErrAlreadyExists` without replacing the original timestamp. Callers can treat replay as success or read the original event. Build IDs are opaque and do not establish transition ordering. The MySQL key uses the same binary identifier limits as the issue stores, plus a 32-byte event field, for a maximum of 2586 bytes. Reads and creates use primary-key operations; there is no update or deletion API, metadata collection, or repair state. -History persistence does not alter open-issue membership or validate transitions against it. No runtime writer is enabled here. A later reconciliation controller must own opening/resolution decisions, retry-safe writes to both stores, and protection against stale build analysis. History enumeration is outside this storage contract. +History persistence does not alter issue state or validate transitions against it. No runtime writer is enabled here. A later reconciliation controller must own opening/resolution decisions, retry-safe writes to both stores, and protection against stale build analysis. History enumeration is outside this storage contract. ## 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/issue_state_store.go b/stovepipe/extension/storage/issue_state_store.go new file mode 100644 index 000000000..2a47bc272 --- /dev/null +++ b/stovepipe/extension/storage/issue_state_store.go @@ -0,0 +1,53 @@ +// 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 storage + +//go:generate mockgen -source=issue_state_store.go -destination=mock/issue_state_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// IssueStateCursor is an exclusive position in bytewise task/unit order. +type IssueStateCursor struct { + // TaskID is the task component; empty with an empty UnitID starts the range. + TaskID string + // UnitID is the unit component; empty identifies a task-wide issue. + UnitID string +} + +// IssueStateRange bounds a primary-key scan within the store's queue. +type IssueStateRange struct { + // After excludes this key and earlier keys; the zero cursor starts the range. + After IssueStateCursor + // Limit is the positive maximum number of records to read, including closed ones. + Limit int +} + +// IssueStateStore retains one versioned record per task/unit within its queue. +// Closing retains the record and its version; there is no deletion operation. +type IssueStateStore interface { + // Create returns ErrAlreadyExists for an existing key, including a closed issue. + Create(context.Context, entity.IssueState) error + // Get returns ErrNotFound only when the task/unit has never been recorded. + Get(ctx context.Context, taskID, unitID string) (entity.IssueState, error) + // Update replaces the record only at oldVersion; callers supply newVersion. + Update(ctx context.Context, issue entity.IssueState, oldVersion, newVersion int32) error + // List returns at most Limit records, including closed issues, in bytewise key order. + // Callers filter by status and continue after the last scanned key, not the last open issue. + List(context.Context, IssueStateRange) ([]entity.IssueState, error) +} diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index 54499c25f..549e2df41 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -6,8 +6,8 @@ go_library( "build_issue_store_mock.go", "build_store_mock.go", "issue_log_store_mock.go", + "issue_state_store_mock.go", "issue_tracking_store_mock.go", - "open_issue_store_mock.go", "project_fact_by_degree_store_mock.go", "queue_log_store_mock.go", "queue_store_mock.go", diff --git a/stovepipe/extension/storage/mock/issue_state_store_mock.go b/stovepipe/extension/storage/mock/issue_state_store_mock.go new file mode 100644 index 000000000..b21be46a4 --- /dev/null +++ b/stovepipe/extension/storage/mock/issue_state_store_mock.go @@ -0,0 +1,101 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: issue_state_store.go +// +// Generated by this command: +// +// mockgen -source=issue_state_store.go -destination=mock/issue_state_store_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/stovepipe/entity" + storage "github.com/uber/submitqueue/stovepipe/extension/storage" + gomock "go.uber.org/mock/gomock" +) + +// MockIssueStateStore is a mock of IssueStateStore interface. +type MockIssueStateStore struct { + ctrl *gomock.Controller + recorder *MockIssueStateStoreMockRecorder + isgomock struct{} +} + +// MockIssueStateStoreMockRecorder is the mock recorder for MockIssueStateStore. +type MockIssueStateStoreMockRecorder struct { + mock *MockIssueStateStore +} + +// NewMockIssueStateStore creates a new mock instance. +func NewMockIssueStateStore(ctrl *gomock.Controller) *MockIssueStateStore { + mock := &MockIssueStateStore{ctrl: ctrl} + mock.recorder = &MockIssueStateStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockIssueStateStore) EXPECT() *MockIssueStateStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockIssueStateStore) Create(arg0 context.Context, arg1 entity.IssueState) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", arg0, arg1) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockIssueStateStoreMockRecorder) Create(arg0, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockIssueStateStore)(nil).Create), arg0, arg1) +} + +// Get mocks base method. +func (m *MockIssueStateStore) Get(ctx context.Context, taskID, unitID string) (entity.IssueState, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, taskID, unitID) + ret0, _ := ret[0].(entity.IssueState) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockIssueStateStoreMockRecorder) Get(ctx, taskID, unitID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockIssueStateStore)(nil).Get), ctx, taskID, unitID) +} + +// List mocks base method. +func (m *MockIssueStateStore) List(arg0 context.Context, arg1 storage.IssueStateRange) ([]entity.IssueState, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "List", arg0, arg1) + ret0, _ := ret[0].([]entity.IssueState) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// List indicates an expected call of List. +func (mr *MockIssueStateStoreMockRecorder) List(arg0, arg1 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockIssueStateStore)(nil).List), arg0, arg1) +} + +// Update mocks base method. +func (m *MockIssueStateStore) Update(ctx context.Context, issue entity.IssueState, oldVersion, newVersion int32) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Update", ctx, issue, oldVersion, newVersion) + ret0, _ := ret[0].(error) + return ret0 +} + +// Update indicates an expected call of Update. +func (mr *MockIssueStateStoreMockRecorder) Update(ctx, issue, oldVersion, newVersion any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*MockIssueStateStore)(nil).Update), ctx, issue, oldVersion, newVersion) +} diff --git a/stovepipe/extension/storage/mock/open_issue_store_mock.go b/stovepipe/extension/storage/mock/open_issue_store_mock.go deleted file mode 100644 index 6df367b63..000000000 --- a/stovepipe/extension/storage/mock/open_issue_store_mock.go +++ /dev/null @@ -1,101 +0,0 @@ -// Code generated by MockGen. DO NOT EDIT. -// Source: open_issue_store.go -// -// Generated by this command: -// -// mockgen -source=open_issue_store.go -destination=mock/open_issue_store_mock.go -package=mock -// - -// Package mock is a generated GoMock package. -package mock - -import ( - context "context" - reflect "reflect" - - entity "github.com/uber/submitqueue/stovepipe/entity" - storage "github.com/uber/submitqueue/stovepipe/extension/storage" - gomock "go.uber.org/mock/gomock" -) - -// MockOpenIssueStore is a mock of OpenIssueStore interface. -type MockOpenIssueStore struct { - ctrl *gomock.Controller - recorder *MockOpenIssueStoreMockRecorder - isgomock struct{} -} - -// MockOpenIssueStoreMockRecorder is the mock recorder for MockOpenIssueStore. -type MockOpenIssueStoreMockRecorder struct { - mock *MockOpenIssueStore -} - -// NewMockOpenIssueStore creates a new mock instance. -func NewMockOpenIssueStore(ctrl *gomock.Controller) *MockOpenIssueStore { - mock := &MockOpenIssueStore{ctrl: ctrl} - mock.recorder = &MockOpenIssueStoreMockRecorder{mock} - return mock -} - -// EXPECT returns an object that allows the caller to indicate expected use. -func (m *MockOpenIssueStore) EXPECT() *MockOpenIssueStoreMockRecorder { - return m.recorder -} - -// Create mocks base method. -func (m *MockOpenIssueStore) Create(ctx context.Context, issue entity.OpenIssue) error { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Create", ctx, issue) - ret0, _ := ret[0].(error) - return ret0 -} - -// Create indicates an expected call of Create. -func (mr *MockOpenIssueStoreMockRecorder) Create(ctx, issue any) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockOpenIssueStore)(nil).Create), ctx, issue) -} - -// Delete mocks base method. -func (m *MockOpenIssueStore) Delete(ctx context.Context, issue entity.OpenIssue) error { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Delete", ctx, issue) - ret0, _ := ret[0].(error) - return ret0 -} - -// Delete indicates an expected call of Delete. -func (mr *MockOpenIssueStoreMockRecorder) Delete(ctx, issue any) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Delete", reflect.TypeOf((*MockOpenIssueStore)(nil).Delete), ctx, issue) -} - -// Get mocks base method. -func (m *MockOpenIssueStore) Get(ctx context.Context, taskID, unitID string) (entity.OpenIssue, error) { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Get", ctx, taskID, unitID) - ret0, _ := ret[0].(entity.OpenIssue) - ret1, _ := ret[1].(error) - return ret0, ret1 -} - -// Get indicates an expected call of Get. -func (mr *MockOpenIssueStoreMockRecorder) Get(ctx, taskID, unitID any) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockOpenIssueStore)(nil).Get), ctx, taskID, unitID) -} - -// List mocks base method. -func (m *MockOpenIssueStore) List(ctx context.Context, bounds storage.OpenIssueRange) ([]entity.OpenIssue, error) { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "List", ctx, bounds) - ret0, _ := ret[0].([]entity.OpenIssue) - ret1, _ := ret[1].(error) - return ret0, ret1 -} - -// List indicates an expected call of List. -func (mr *MockOpenIssueStoreMockRecorder) List(ctx, bounds any) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockOpenIssueStore)(nil).List), ctx, bounds) -} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index 7b1e2c61d..ec86a0158 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -121,32 +121,32 @@ func (mr *MockStorageMockRecorder) GetIssueLogStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetIssueLogStore", reflect.TypeOf((*MockStorage)(nil).GetIssueLogStore)) } -// GetIssueTrackingStore mocks base method. -func (m *MockStorage) GetIssueTrackingStore() storage.IssueTrackingStore { +// GetIssueStateStore mocks base method. +func (m *MockStorage) GetIssueStateStore() storage.IssueStateStore { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetIssueTrackingStore") - ret0, _ := ret[0].(storage.IssueTrackingStore) + ret := m.ctrl.Call(m, "GetIssueStateStore") + ret0, _ := ret[0].(storage.IssueStateStore) return ret0 } -// GetIssueTrackingStore indicates an expected call of GetIssueTrackingStore. -func (mr *MockStorageMockRecorder) GetIssueTrackingStore() *gomock.Call { +// GetIssueStateStore indicates an expected call of GetIssueStateStore. +func (mr *MockStorageMockRecorder) GetIssueStateStore() *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetIssueTrackingStore", reflect.TypeOf((*MockStorage)(nil).GetIssueTrackingStore)) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetIssueStateStore", reflect.TypeOf((*MockStorage)(nil).GetIssueStateStore)) } -// GetOpenIssueStore mocks base method. -func (m *MockStorage) GetOpenIssueStore() storage.OpenIssueStore { +// GetIssueTrackingStore mocks base method. +func (m *MockStorage) GetIssueTrackingStore() storage.IssueTrackingStore { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetOpenIssueStore") - ret0, _ := ret[0].(storage.OpenIssueStore) + ret := m.ctrl.Call(m, "GetIssueTrackingStore") + ret0, _ := ret[0].(storage.IssueTrackingStore) return ret0 } -// GetOpenIssueStore indicates an expected call of GetOpenIssueStore. -func (mr *MockStorageMockRecorder) GetOpenIssueStore() *gomock.Call { +// GetIssueTrackingStore indicates an expected call of GetIssueTrackingStore. +func (mr *MockStorageMockRecorder) GetIssueTrackingStore() *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetOpenIssueStore", reflect.TypeOf((*MockStorage)(nil).GetOpenIssueStore)) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetIssueTrackingStore", reflect.TypeOf((*MockStorage)(nil).GetIssueTrackingStore)) } // GetProjectFactByDegreeStore mocks base method. diff --git a/stovepipe/extension/storage/mysql/BUILD.bazel b/stovepipe/extension/storage/mysql/BUILD.bazel index 9ffd7d4eb..f5660c3fc 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/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", @@ -39,9 +39,9 @@ go_test( "build_issue_update_test.go", "build_store_test.go", "issue_log_store_test.go", + "issue_state_list_test.go", + "issue_state_store_test.go", "issue_tracking_store_test.go", - "open_issue_list_test.go", - "open_issue_store_test.go", "project_fact_by_degree_store_test.go", "queue_store_test.go", "queue_summary_store_test.go", diff --git a/stovepipe/extension/storage/mysql/open_issue_list_test.go b/stovepipe/extension/storage/mysql/issue_state_list_test.go similarity index 63% rename from stovepipe/extension/storage/mysql/open_issue_list_test.go rename to stovepipe/extension/storage/mysql/issue_state_list_test.go index 01d75343c..e2afedebb 100644 --- a/stovepipe/extension/storage/mysql/open_issue_list_test.go +++ b/stovepipe/extension/storage/mysql/issue_state_list_test.go @@ -28,20 +28,20 @@ import ( "github.com/uber/submitqueue/stovepipe/extension/storage" ) -func TestOpenIssueList(t *testing.T) { +func TestIssueStateList(t *testing.T) { for _, tt := range []struct { name string - after storage.OpenIssueCursor + after storage.IssueStateCursor empty bool dbErr error scanError bool rowError bool }{ {name: "first page"}, - {name: "after task-wide issue", after: storage.OpenIssueCursor{TaskID: "task"}}, + {name: "after task-wide issue", after: storage.IssueStateCursor{TaskID: "task"}}, {name: "empty page", empty: true}, {name: "query error", dbErr: assert.AnError}, - {name: "cancelled", dbErr: context.Canceled}, + {name: "canceled", dbErr: context.Canceled}, {name: "scan error", scanError: true}, {name: "iteration error", rowError: true}, } { @@ -49,33 +49,33 @@ func TestOpenIssueList(t *testing.T) { db, mock, err := sqlmock.New() require.NoError(t, err) defer db.Close() - store := NewOpenIssueStore(db, testMetrics(), "queue") - issue := openIssueFixture() - query := listOpenIssuesQuery + store := NewIssueStateStore(db, testMetrics(), "queue") + issue := issueStateFixture() + query := listIssueStatesQuery args := []driver.Value{"queue", int64(2)} if tt.after.TaskID != "" { - query = listOpenIssuesAfterCursorQuery + query = listIssueStatesAfterCursorQuery args = []driver.Value{"queue", tt.after.TaskID, tt.after.TaskID, tt.after.UnitID, int64(2)} } expectation := mock.ExpectQuery(regexp.QuoteMeta(query)).WithArgs(args...) if tt.dbErr != nil { expectation.WillReturnError(tt.dbErr) } else { - rows := sqlmock.NewRows([]string{"task_id", "unit_id", "first_seen_build_id"}) + rows := sqlmock.NewRows([]string{"task_id", "unit_id", "status", "first_seen_build_id", "version"}) if !tt.empty { var firstBuild any = issue.FirstSeenBuildID if tt.scanError { firstBuild = nil } - rows.AddRow(issue.TaskID, issue.UnitID, firstBuild) + rows.AddRow(issue.TaskID, issue.UnitID, issue.Status, firstBuild, issue.Version) if tt.rowError { - rows.AddRow("task", "unit-2", "build/2").RowError(1, assert.AnError) + rows.AddRow("task", "unit-2", entity.IssueStatusClosed, "build/2", 2).RowError(1, assert.AnError) } } expectation.WillReturnRows(rows).RowsWillBeClosed() } - got, err := store.List(context.Background(), storage.OpenIssueRange{After: tt.after, Limit: 2}) + got, err := store.List(context.Background(), storage.IssueStateRange{After: tt.after, Limit: 2}) switch { case tt.dbErr != nil: require.ErrorIs(t, err, tt.dbErr) @@ -92,28 +92,28 @@ func TestOpenIssueList(t *testing.T) { require.Empty(t, got) default: require.NoError(t, err) - require.Equal(t, []entity.OpenIssue{issue}, got) + require.Equal(t, []entity.IssueState{issue}, got) } require.NoError(t, mock.ExpectationsWereMet()) }) } } -func TestOpenIssueListRejectsInvalidRange(t *testing.T) { +func TestIssueStateListRejectsInvalidRange(t *testing.T) { for _, tt := range []struct { name string queue string - bounds storage.OpenIssueRange + bounds storage.IssueStateRange }{ - {name: "zero limit", bounds: storage.OpenIssueRange{}}, - {name: "negative limit", bounds: storage.OpenIssueRange{Limit: -1}}, - {name: "unit without task", bounds: storage.OpenIssueRange{Limit: 1, - After: storage.OpenIssueCursor{UnitID: "unit"}}}, - {name: "oversized task", bounds: storage.OpenIssueRange{Limit: 1, - After: storage.OpenIssueCursor{TaskID: strings.Repeat("t", 256)}}}, - {name: "oversized unit", bounds: storage.OpenIssueRange{Limit: 1, - After: storage.OpenIssueCursor{TaskID: "task", UnitID: strings.Repeat("😀", 257)}}}, - {name: "oversized queue", queue: strings.Repeat("q", 256), bounds: storage.OpenIssueRange{Limit: 1}}, + {name: "zero limit", bounds: storage.IssueStateRange{}}, + {name: "negative limit", bounds: storage.IssueStateRange{Limit: -1}}, + {name: "unit without task", bounds: storage.IssueStateRange{Limit: 1, + After: storage.IssueStateCursor{UnitID: "unit"}}}, + {name: "oversized task", bounds: storage.IssueStateRange{Limit: 1, + After: storage.IssueStateCursor{TaskID: strings.Repeat("t", 256)}}}, + {name: "oversized unit", bounds: storage.IssueStateRange{Limit: 1, + After: storage.IssueStateCursor{TaskID: "task", UnitID: strings.Repeat("😀", 257)}}}, + {name: "oversized queue", queue: strings.Repeat("q", 256), bounds: storage.IssueStateRange{Limit: 1}}, } { t.Run(tt.name, func(t *testing.T) { db, mock, err := sqlmock.New() @@ -123,8 +123,8 @@ func TestOpenIssueListRejectsInvalidRange(t *testing.T) { if queue == "" { queue = "queue" } - store := NewOpenIssueStore(db, testMetrics(), queue) - mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "first_seen_build_id"})) + store := NewIssueStateStore(db, testMetrics(), queue) + mock.ExpectQuery("SELECT").WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "status", "first_seen_build_id", "version"})) got, err := store.List(context.Background(), tt.bounds) require.Error(t, err) diff --git a/stovepipe/extension/storage/mysql/issue_state_store.go b/stovepipe/extension/storage/mysql/issue_state_store.go new file mode 100644 index 000000000..8e9a2671c --- /dev/null +++ b/stovepipe/extension/storage/mysql/issue_state_store.go @@ -0,0 +1,183 @@ +// 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 mysql + +import ( + "context" + "database/sql" + "errors" + "fmt" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +const listIssueStatesQuery = ` + SELECT task_id, unit_id, status, first_seen_build_id, version + FROM issue_state WHERE queue = ? + ORDER BY task_id ASC, unit_id ASC LIMIT ?` + +const listIssueStatesAfterCursorQuery = ` + SELECT task_id, unit_id, status, first_seen_build_id, version + FROM issue_state WHERE queue = ? + AND (task_id > ? OR (task_id = ? AND unit_id > ?)) + ORDER BY task_id ASC, unit_id ASC LIMIT ?` + +type issueStateStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewIssueStateStore binds retained issue states to one queue. +func NewIssueStateStore(db *sql.DB, scope tally.Scope, queue string) storage.IssueStateStore { + return &issueStateStore{db: db, scope: scope, queue: queue} +} + +func (s *issueStateStore) Create(ctx context.Context, issue entity.IssueState) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := s.validateIssue(issue); err != nil { + return err + } + + _, err := s.db.ExecContext(ctx, + `INSERT INTO issue_state (queue, task_id, unit_id, status, first_seen_build_id, version) + VALUES (?, ?, ?, ?, ?, ?)`, + s.queue, issue.TaskID, issue.UnitID, issue.Status, issue.FirstSeenBuildID, issue.Version) + if err != nil { + if isDuplicateEntry(err) { + return fmt.Errorf("issue state task=%q unit=%q: %w", issue.TaskID, issue.UnitID, storage.ErrAlreadyExists) + } + + return fmt.Errorf("failed to create issue state: %w", err) + } + + return nil +} + +func (s *issueStateStore) Get(ctx context.Context, taskID, unitID string) (ret entity.IssueState, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if taskID == "" { + return entity.IssueState{}, fmt.Errorf("issue state task ID must not be empty") + } + if err := validateBuildIssueKey(s.queue, "", taskID, unitID); err != nil { + return entity.IssueState{}, err + } + + var issue entity.IssueState + err := s.db.QueryRowContext(ctx, + `SELECT task_id, unit_id, status, first_seen_build_id, version + FROM issue_state WHERE queue = ? AND task_id = ? AND unit_id = ?`, + s.queue, taskID, unitID, + ).Scan(&issue.TaskID, &issue.UnitID, &issue.Status, &issue.FirstSeenBuildID, &issue.Version) + if errors.Is(err, sql.ErrNoRows) { + return entity.IssueState{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.IssueState{}, fmt.Errorf("failed to get issue state: %w", err) + } + + return issue, nil +} + +func (s *issueStateStore) List(ctx context.Context, bounds storage.IssueStateRange) (ret []entity.IssueState, retErr error) { + op := metrics.Begin(s.scope, "list", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + cursor := bounds.After + if bounds.Limit <= 0 { + return nil, fmt.Errorf("issue state range requires a positive limit") + } + if cursor.TaskID == "" && cursor.UnitID != "" { + return nil, fmt.Errorf("issue state cursor requires a task ID when a unit ID is supplied") + } + if err := validateBuildIssueKey(s.queue, "", cursor.TaskID, cursor.UnitID); err != nil { + return nil, err + } + + var ( + rows *sql.Rows + err error + ) + if cursor.TaskID != "" { + rows, err = s.db.QueryContext(ctx, listIssueStatesAfterCursorQuery, + s.queue, cursor.TaskID, cursor.TaskID, cursor.UnitID, bounds.Limit) + } else { + rows, err = s.db.QueryContext(ctx, listIssueStatesQuery, s.queue, bounds.Limit) + } + if err != nil { + return nil, fmt.Errorf("failed to list issue states: %w", err) + } + defer rows.Close() + + issues := make([]entity.IssueState, 0) + for rows.Next() { + var issue entity.IssueState + if err := rows.Scan(&issue.TaskID, &issue.UnitID, &issue.Status, &issue.FirstSeenBuildID, &issue.Version); err != nil { + return nil, fmt.Errorf("failed to scan issue state: %w", err) + } + issues = append(issues, issue) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to iterate issue states: %w", err) + } + + return issues, nil +} + +func (s *issueStateStore) Update(ctx context.Context, issue entity.IssueState, oldVersion, newVersion int32) (retErr error) { + op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + replacement := issue + replacement.Version = newVersion + if err := s.validateIssue(replacement); err != nil { + return err + } + if oldVersion <= 0 { + return fmt.Errorf("issue state guard version must be positive") + } + result, err := s.db.ExecContext(ctx, + `UPDATE issue_state SET status = ?, first_seen_build_id = ?, version = ? WHERE queue = ? AND task_id = ? AND unit_id = ? AND version = ?`, + issue.Status, issue.FirstSeenBuildID, newVersion, s.queue, issue.TaskID, issue.UnitID, oldVersion) + if err != nil { + return fmt.Errorf("update issue state: %w", err) + } + rows, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("read issue state update count: %w", err) + } + if rows != 1 { + return storage.ErrVersionMismatch + } + return nil +} + +func (s *issueStateStore) validateIssue(issue entity.IssueState) error { + if err := issue.Validate(); err != nil { + return err + } + if err := validateBuildIssueKey(s.queue, issue.FirstSeenBuildID, issue.TaskID, issue.UnitID); err != nil { + return err + } + return nil +} + +var _ storage.IssueStateStore = (*issueStateStore)(nil) diff --git a/stovepipe/extension/storage/mysql/open_issue_store_test.go b/stovepipe/extension/storage/mysql/issue_state_store_test.go similarity index 57% rename from stovepipe/extension/storage/mysql/open_issue_store_test.go rename to stovepipe/extension/storage/mysql/issue_state_store_test.go index 9fd96051f..b17551aaf 100644 --- a/stovepipe/extension/storage/mysql/open_issue_store_test.go +++ b/stovepipe/extension/storage/mysql/issue_state_store_test.go @@ -29,14 +29,14 @@ import ( "github.com/uber/submitqueue/stovepipe/extension/storage" ) -func openIssueFixture() entity.OpenIssue { - return entity.OpenIssue{ +func issueStateFixture() entity.IssueState { + return entity.IssueState{ TaskID: "task", UnitID: "unit", - FirstSeenBuildID: "build/1", + FirstSeenBuildID: "build/1", Status: entity.IssueStatusOpen, Version: 1, } } -func TestOpenIssueCreate(t *testing.T) { +func TestIssueStateCreate(t *testing.T) { for _, tt := range []struct { name string dbErr error @@ -50,13 +50,13 @@ func TestOpenIssueCreate(t *testing.T) { db, mock, err := sqlmock.New() require.NoError(t, err) defer db.Close() - issue := openIssueFixture() - store := NewOpenIssueStore(db, testMetrics(), "queue") + issue := issueStateFixture() + store := NewIssueStateStore(db, testMetrics(), "queue") q := mock.ExpectExec(regexp.QuoteMeta( - `INSERT INTO open_issue (queue, task_id, unit_id, first_seen_build_id) - VALUES (?, ?, ?, ?)`)). - WithArgs("queue", issue.TaskID, issue.UnitID, issue.FirstSeenBuildID) + `INSERT INTO issue_state (queue, task_id, unit_id, status, first_seen_build_id, version) + VALUES (?, ?, ?, ?, ?, ?)`)). + WithArgs("queue", issue.TaskID, issue.UnitID, issue.Status, issue.FirstSeenBuildID, issue.Version) if tt.dbErr != nil { q.WillReturnError(tt.dbErr) } else { @@ -75,18 +75,18 @@ func TestOpenIssueCreate(t *testing.T) { } } -func TestOpenIssueGet(t *testing.T) { +func TestIssueStateGet(t *testing.T) { for _, kind := range []string{"found", "missing", "database error", "scan error"} { t.Run(kind, func(t *testing.T) { db, mock, err := sqlmock.New() require.NoError(t, err) defer db.Close() - issue := openIssueFixture() - store := NewOpenIssueStore(db, testMetrics(), "queue") + issue := issueStateFixture() + store := NewIssueStateStore(db, testMetrics(), "queue") q := mock.ExpectQuery(regexp.QuoteMeta( - `SELECT task_id, unit_id, first_seen_build_id - FROM open_issue WHERE queue = ? AND task_id = ? AND unit_id = ?`)). + `SELECT task_id, unit_id, status, first_seen_build_id, version + FROM issue_state WHERE queue = ? AND task_id = ? AND unit_id = ?`)). WithArgs("queue", issue.TaskID, issue.UnitID) switch kind { case "missing": @@ -99,8 +99,8 @@ func TestOpenIssueGet(t *testing.T) { firstBuild = nil } - q.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "first_seen_build_id"}). - AddRow(issue.TaskID, issue.UnitID, firstBuild)) + q.WillReturnRows(sqlmock.NewRows([]string{"task_id", "unit_id", "status", "first_seen_build_id", "version"}). + AddRow(issue.TaskID, issue.UnitID, issue.Status, firstBuild, issue.Version)) } got, err := store.Get(context.Background(), issue.TaskID, issue.UnitID) @@ -117,7 +117,7 @@ func TestOpenIssueGet(t *testing.T) { } if kind != "found" { - assert.Equal(t, entity.OpenIssue{}, got) + assert.Equal(t, entity.IssueState{}, got) } require.NoError(t, mock.ExpectationsWereMet()) @@ -125,54 +125,55 @@ func TestOpenIssueGet(t *testing.T) { } } -func TestOpenIssueDelete(t *testing.T) { +func TestIssueStateUpdate(t *testing.T) { for _, tt := range []struct { - name string - rows int64 - dbErr error + name string + rows int64 + dbErr, wantErr error + resultError bool }{ - {name: "matching opening", rows: 1}, - {name: "already absent"}, - {name: "different opening"}, - {name: "database error", dbErr: assert.AnError}, + {name: "updated", rows: 1}, + {name: "stale version", wantErr: storage.ErrVersionMismatch}, + {name: "database error", dbErr: assert.AnError, wantErr: assert.AnError}, + {name: "result error", resultError: true, wantErr: assert.AnError}, } { t.Run(tt.name, func(t *testing.T) { db, mock, err := sqlmock.New() require.NoError(t, err) defer db.Close() - issue := openIssueFixture() - store := NewOpenIssueStore(db, testMetrics(), "queue") - - q := mock.ExpectExec(regexp.QuoteMeta( - `DELETE FROM open_issue - WHERE queue = ? AND task_id = ? AND unit_id = ? AND first_seen_build_id = ?`)). - WithArgs("queue", issue.TaskID, issue.UnitID, issue.FirstSeenBuildID) - if tt.dbErr != nil { - q.WillReturnError(tt.dbErr) - } else { - q.WillReturnResult(sqlmock.NewResult(0, tt.rows)) + issue := issueStateFixture() + issue.Status = entity.IssueStatusClosed + expectation := mock.ExpectExec(regexp.QuoteMeta(`UPDATE issue_state SET status = ?, first_seen_build_id = ?, version = ? WHERE queue = ? AND task_id = ? AND unit_id = ? AND version = ?`)). + WithArgs(issue.Status, issue.FirstSeenBuildID, int32(2), "queue", issue.TaskID, issue.UnitID, int32(1)) + switch { + case tt.dbErr != nil: + expectation.WillReturnError(tt.dbErr) + case tt.resultError: + expectation.WillReturnResult(sqlmock.NewErrorResult(assert.AnError)) + default: + expectation.WillReturnResult(sqlmock.NewResult(0, tt.rows)) } - - err = store.Delete(context.Background(), issue) - if tt.dbErr != nil { - require.ErrorIs(t, err, tt.dbErr) + before := issue + err = NewIssueStateStore(db, testMetrics(), "queue").Update(context.Background(), issue, 1, 2) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) } else { require.NoError(t, err) } - + require.Equal(t, before, issue) require.NoError(t, mock.ExpectationsWereMet()) }) } } -func TestOpenIssueRejectsInvalidIdentifiers(t *testing.T) { - for _, kind := range []string{"queue", "task", "empty task", "unit", "first build", "empty first build"} { +func TestIssueStateRejectsInvalidIdentifiers(t *testing.T) { + for _, kind := range []string{"queue", "task", "empty task", "unit", "first build", "empty first build", "unknown status", "zero version"} { t.Run(kind, func(t *testing.T) { db, mock, err := sqlmock.New() require.NoError(t, err) defer db.Close() queue := "queue" - issue := openIssueFixture() + issue := issueStateFixture() switch kind { case "queue": @@ -187,19 +188,23 @@ func TestOpenIssueRejectsInvalidIdentifiers(t *testing.T) { issue.FirstSeenBuildID = strings.Repeat("b", 1021) case "empty first build": issue.FirstSeenBuildID = "" + case "unknown status": + issue.Status = entity.IssueStatusUnknown + case "zero version": + issue.Version = 0 } mock.MatchExpectationsInOrder(false) - mock.ExpectExec("INSERT INTO open_issue").WillReturnResult(sqlmock.NewResult(0, 1)) - mock.ExpectExec("DELETE FROM open_issue").WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec("INSERT INTO issue_state").WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec("UPDATE issue_state").WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectQuery("SELECT task_id, unit_id").WillReturnRows( - sqlmock.NewRows([]string{"task_id", "unit_id", "first_seen_build_id"}). - AddRow(issue.TaskID, issue.UnitID, issue.FirstSeenBuildID), + sqlmock.NewRows([]string{"task_id", "unit_id", "status", "first_seen_build_id", "version"}). + AddRow(issue.TaskID, issue.UnitID, issue.Status, issue.FirstSeenBuildID, issue.Version), ) - store := NewOpenIssueStore(db, testMetrics(), queue) + store := NewIssueStateStore(db, testMetrics(), queue) require.Error(t, store.Create(context.Background(), issue)) - require.Error(t, store.Delete(context.Background(), issue)) + require.Error(t, store.Update(context.Background(), issue, 1, issue.Version)) switch kind { case "queue", "task", "empty task", "unit": 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/open_issue_store.go b/stovepipe/extension/storage/mysql/open_issue_store.go deleted file mode 100644 index ecf16efb8..000000000 --- a/stovepipe/extension/storage/mysql/open_issue_store.go +++ /dev/null @@ -1,176 +0,0 @@ -// 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 mysql - -import ( - "context" - "database/sql" - "errors" - "fmt" - - "github.com/uber-go/tally" - "github.com/uber/submitqueue/platform/metrics" - "github.com/uber/submitqueue/stovepipe/entity" - "github.com/uber/submitqueue/stovepipe/extension/storage" -) - -const listOpenIssuesQuery = ` - SELECT task_id, unit_id, first_seen_build_id - FROM open_issue WHERE queue = ? - ORDER BY task_id ASC, unit_id ASC LIMIT ?` - -const listOpenIssuesAfterCursorQuery = ` - SELECT task_id, unit_id, first_seen_build_id - FROM open_issue WHERE queue = ? - AND (task_id > ? OR (task_id = ? AND unit_id > ?)) - ORDER BY task_id ASC, unit_id ASC LIMIT ?` - -type openIssueStore struct { - db *sql.DB - scope tally.Scope - queue string -} - -// NewOpenIssueStore binds currently open issues to queue over the shared database. -func NewOpenIssueStore(db *sql.DB, scope tally.Scope, queue string) storage.OpenIssueStore { - return &openIssueStore{db: db, scope: scope, queue: queue} -} - -func (s *openIssueStore) Create(ctx context.Context, issue entity.OpenIssue) (retErr error) { - op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) - defer func() { op.Complete(retErr) }() - - if err := s.validateIssue(issue); err != nil { - return err - } - - _, err := s.db.ExecContext(ctx, - `INSERT INTO open_issue (queue, task_id, unit_id, first_seen_build_id) - VALUES (?, ?, ?, ?)`, - s.queue, issue.TaskID, issue.UnitID, issue.FirstSeenBuildID) - if err != nil { - if isDuplicateEntry(err) { - return fmt.Errorf("open issue task=%q unit=%q: %w", issue.TaskID, issue.UnitID, storage.ErrAlreadyExists) - } - - return fmt.Errorf("failed to create open issue: %w", err) - } - - return nil -} - -func (s *openIssueStore) Get(ctx context.Context, taskID, unitID string) (ret entity.OpenIssue, retErr error) { - op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) - defer func() { op.Complete(retErr) }() - - if taskID == "" { - return entity.OpenIssue{}, fmt.Errorf("open issue task ID must not be empty") - } - if err := validateBuildIssueKey(s.queue, "", taskID, unitID); err != nil { - return entity.OpenIssue{}, err - } - - var issue entity.OpenIssue - err := s.db.QueryRowContext(ctx, - `SELECT task_id, unit_id, first_seen_build_id - FROM open_issue WHERE queue = ? AND task_id = ? AND unit_id = ?`, - s.queue, taskID, unitID, - ).Scan(&issue.TaskID, &issue.UnitID, &issue.FirstSeenBuildID) - if errors.Is(err, sql.ErrNoRows) { - return entity.OpenIssue{}, storage.WrapNotFound(err) - } - if err != nil { - return entity.OpenIssue{}, fmt.Errorf("failed to get open issue: %w", err) - } - - return issue, nil -} - -func (s *openIssueStore) List(ctx context.Context, bounds storage.OpenIssueRange) (ret []entity.OpenIssue, retErr error) { - op := metrics.Begin(s.scope, "list", metrics.StorageLatencyBuckets) - defer func() { op.Complete(retErr) }() - - cursor := bounds.After - if bounds.Limit <= 0 { - return nil, fmt.Errorf("open issue range requires a positive limit") - } - if cursor.TaskID == "" && cursor.UnitID != "" { - return nil, fmt.Errorf("open issue cursor requires a task ID when a unit ID is supplied") - } - if err := validateBuildIssueKey(s.queue, "", cursor.TaskID, cursor.UnitID); err != nil { - return nil, err - } - - var ( - rows *sql.Rows - err error - ) - if cursor.TaskID != "" { - rows, err = s.db.QueryContext(ctx, listOpenIssuesAfterCursorQuery, - s.queue, cursor.TaskID, cursor.TaskID, cursor.UnitID, bounds.Limit) - } else { - rows, err = s.db.QueryContext(ctx, listOpenIssuesQuery, s.queue, bounds.Limit) - } - if err != nil { - return nil, fmt.Errorf("failed to list open issues: %w", err) - } - defer rows.Close() - - issues := make([]entity.OpenIssue, 0) - for rows.Next() { - var issue entity.OpenIssue - if err := rows.Scan(&issue.TaskID, &issue.UnitID, &issue.FirstSeenBuildID); err != nil { - return nil, fmt.Errorf("failed to scan open issue: %w", err) - } - issues = append(issues, issue) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("failed to iterate open issues: %w", err) - } - - return issues, nil -} - -func (s *openIssueStore) Delete(ctx context.Context, issue entity.OpenIssue) (retErr error) { - op := metrics.Begin(s.scope, "delete", metrics.StorageLatencyBuckets) - defer func() { op.Complete(retErr) }() - - if err := s.validateIssue(issue); err != nil { - return err - } - - _, err := s.db.ExecContext(ctx, - `DELETE FROM open_issue - WHERE queue = ? AND task_id = ? AND unit_id = ? AND first_seen_build_id = ?`, - s.queue, issue.TaskID, issue.UnitID, issue.FirstSeenBuildID) - if err != nil { - return fmt.Errorf("failed to delete open issue: %w", err) - } - - return nil -} - -func (s *openIssueStore) validateIssue(issue entity.OpenIssue) error { - if issue.TaskID == "" { - return fmt.Errorf("open issue task ID must not be empty") - } - if err := validateBuildIssueKey(s.queue, issue.FirstSeenBuildID, issue.TaskID, issue.UnitID); err != nil { - return err - } - if issue.FirstSeenBuildID == "" { - return fmt.Errorf("open issue first-seen build ID must not be empty") - } - return nil -} diff --git a/stovepipe/extension/storage/mysql/schema/open_issue.sql b/stovepipe/extension/storage/mysql/schema/issue_state.sql similarity index 60% rename from stovepipe/extension/storage/mysql/schema/open_issue.sql rename to stovepipe/extension/storage/mysql/schema/issue_state.sql index d42436258..5ee7c5283 100644 --- a/stovepipe/extension/storage/mysql/schema/open_issue.sql +++ b/stovepipe/extension/storage/mysql/schema/issue_state.sql @@ -1,9 +1,12 @@ --- Only currently open issues; historical observations remain in build_issue. -CREATE TABLE IF NOT EXISTS open_issue ( +-- Closed records retain their versions so stale writes cannot recreate old openings. +CREATE TABLE IF NOT EXISTS issue_state ( queue VARBINARY(255) NOT NULL, task_id VARBINARY(255) NOT NULL, unit_id VARBINARY(1024) NOT NULL, + status VARBINARY(32) NOT NULL, + version INT NOT NULL, + first_seen_build_id VARBINARY(1020) NOT NULL, -- Accommodates existing VARCHAR(255) utf8mb4 build IDs. PRIMARY KEY (queue, task_id, unit_id) 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/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go index 4a00fa56c..afb28d28e 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -56,7 +56,7 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { queueStore: NewQueueStore(s.db, queueScope.SubScope("queue_store"), queueName), buildStore: NewBuildStore(s.db, queueScope.SubScope("build_store"), queueName), buildIssueStore: NewBuildIssueStore(s.db, queueScope.SubScope("build_issue_store"), queueName), - openIssueStore: NewOpenIssueStore(s.db, queueScope.SubScope("open_issue_store"), queueName), + issueStateStore: NewIssueStateStore(s.db, queueScope.SubScope("issue_state_store"), queueName), issueLogStore: NewIssueLogStore(s.db, queueScope.SubScope("issue_log_store"), queueName), issueTrackingStore: NewIssueTrackingStore(s.db, queueScope.SubScope("issue_tracking_store"), queueName), validationFactStore: NewValidationFactStore(s.db, queueScope.SubScope("validation_fact_store"), queueName), @@ -81,7 +81,7 @@ type mysqlStorage struct { queueLogStore storage.QueueLogStore buildStore storage.BuildStore buildIssueStore storage.BuildIssueStore - openIssueStore storage.OpenIssueStore + issueStateStore storage.IssueStateStore issueLogStore storage.IssueLogStore issueTrackingStore storage.IssueTrackingStore validationFactStore storage.ValidationFactStore @@ -145,8 +145,8 @@ func (f *mysqlStorage) GetBuildIssueStore() storage.BuildIssueStore { return f.buildIssueStore } -func (f *mysqlStorage) GetOpenIssueStore() storage.OpenIssueStore { - return f.openIssueStore +func (f *mysqlStorage) GetIssueStateStore() storage.IssueStateStore { + return f.issueStateStore } func (f *mysqlStorage) GetIssueLogStore() storage.IssueLogStore { diff --git a/stovepipe/extension/storage/mysql/storage_test.go b/stovepipe/extension/storage/mysql/storage_test.go index 291bf9697..a8e7a4f4b 100644 --- a/stovepipe/extension/storage/mysql/storage_test.go +++ b/stovepipe/extension/storage/mysql/storage_test.go @@ -49,7 +49,7 @@ func TestNewStorage(t *testing.T) { assert.NotNil(t, bound.GetQueueStore()) assert.NotNil(t, bound.GetBuildStore()) assert.NotNil(t, bound.GetBuildIssueStore()) - assert.NotNil(t, bound.GetOpenIssueStore()) + assert.NotNil(t, bound.GetIssueStateStore()) assert.NotNil(t, bound.GetIssueLogStore()) assert.NotNil(t, bound.GetIssueTrackingStore()) @@ -141,10 +141,10 @@ func TestStorageForQueueMetrics(t *testing.T) { }, }, { - name: "open_issue_store", + name: "issue_state_store", op: "get", read: func(bound storage.Storage, _ string) error { - _, err := bound.GetOpenIssueStore().Get(context.Background(), "task-id", "unit-id") + _, err := bound.GetIssueStateStore().Get(context.Background(), "task-id", "unit-id") return err }, }, diff --git a/stovepipe/extension/storage/open_issue_store.go b/stovepipe/extension/storage/open_issue_store.go deleted file mode 100644 index a5231397d..000000000 --- a/stovepipe/extension/storage/open_issue_store.go +++ /dev/null @@ -1,60 +0,0 @@ -// 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 storage - -//go:generate mockgen -source=open_issue_store.go -destination=mock/open_issue_store_mock.go -package=mock - -import ( - "context" - - "github.com/uber/submitqueue/stovepipe/entity" -) - -// OpenIssueCursor is an exclusive position in ascending bytewise task/unit order. -type OpenIssueCursor struct { - // TaskID is the task component; empty with an empty UnitID starts the range. - TaskID string - // UnitID is the unit component of the last returned key; empty is a task-wide issue. - UnitID string -} - -// OpenIssueRange bounds a primary-key scan within the store's queue. -type OpenIssueRange struct { - // After excludes this key and earlier keys; the zero cursor starts the range. - After OpenIssueCursor - // Limit is the positive maximum number of issues to return. - Limit int -} - -// OpenIssueStore contains only currently open issues, keyed by task and unit -// within the bound queue. A reopened issue must use a new first-seen build ID. -// Issue keys require a nonempty task ID; an empty unit is a task-wide issue. -type OpenIssueStore interface { - // Create requires a nonempty first-seen build ID. - // An existing task/unit returns ErrAlreadyExists. - Create(ctx context.Context, issue entity.OpenIssue) error - - // Get returns ErrNotFound when the task/unit is not currently open. - Get(ctx context.Context, taskID, unitID string) (entity.OpenIssue, error) - - // List returns at most Limit issues in ascending bytewise (task, unit) order. - // An empty slice ends the range; invalid bounds return an error. - // Pages are not a snapshot: concurrent creates/deletes may change membership. - List(ctx context.Context, bounds OpenIssueRange) ([]entity.OpenIssue, error) - - // Delete removes only the matching opening. It succeeds if that opening is - // already absent and never removes a row with a different first-seen build. - Delete(ctx context.Context, issue entity.OpenIssue) error -} diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index 90d4d55a5..68132fd04 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -97,8 +97,8 @@ type Storage interface { // GetBuildIssueStore returns the BuildIssueStore instance. GetBuildIssueStore() BuildIssueStore - // GetOpenIssueStore returns the OpenIssueStore instance. - GetOpenIssueStore() OpenIssueStore + // GetIssueStateStore returns the IssueStateStore instance. + GetIssueStateStore() IssueStateStore // GetIssueLogStore returns the IssueLogStore instance. GetIssueLogStore() IssueLogStore diff --git a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index eca2e2760..e4a69b5aa 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -7,9 +7,9 @@ go_test( "build_issue_test.go", "build_issue_update_test.go", "issue_log_test.go", + "issue_state_list_test.go", + "issue_state_test.go", "issue_tracking_test.go", - "open_issue_list_test.go", - "open_issue_test.go", "project_fact_by_degree_test.go", "queue_state_transitions_test.go", "queue_summary_test.go", diff --git a/test/integration/stovepipe/extension/storage/mysql/build_issue_update_test.go b/test/integration/stovepipe/extension/storage/mysql/build_issue_update_test.go index 304afd3a9..dcde33adb 100644 --- a/test/integration/stovepipe/extension/storage/mysql/build_issue_update_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/build_issue_update_test.go @@ -44,9 +44,10 @@ func testBuildIssueUpdate(t *testing.T, ctx context.Context, db *sql.DB, factory require.ErrorIs(t, store.Update(ctx, "Build ", issue, 0, 1), storage.ErrVersionMismatch) require.ErrorIs(t, store.Create(ctx, "Build ", issue), storage.ErrAlreadyExists) - open := entity.OpenIssue{TaskID: issue.TaskID, UnitID: issue.UnitID, FirstSeenBuildID: "Build "} - require.NoError(t, bound.GetOpenIssueStore().Create(ctx, open)) - require.NoError(t, bound.GetOpenIssueStore().Delete(ctx, open)) + open := entity.IssueState{TaskID: issue.TaskID, UnitID: issue.UnitID, FirstSeenBuildID: "Build ", Status: entity.IssueStatusOpen, Version: 1} + require.NoError(t, bound.GetIssueStateStore().Create(ctx, open)) + open.Status = entity.IssueStatusClosed + require.NoError(t, bound.GetIssueStateStore().Update(ctx, open, 1, 2)) page, err := store.List(ctx, "Build ", storage.BuildIssueRange{Limit: 1}) require.NoError(t, err) require.Equal(t, []entity.BuildIssue{updated}, page) diff --git a/test/integration/stovepipe/extension/storage/mysql/issue_log_test.go b/test/integration/stovepipe/extension/storage/mysql/issue_log_test.go index 50dbf0b59..6d14dbbbc 100644 --- a/test/integration/stovepipe/extension/storage/mysql/issue_log_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/issue_log_test.go @@ -34,23 +34,26 @@ func testIssueLogStore(t *testing.T, ctx context.Context, factory storage.Factor opened := entity.IssueLog{TaskID: "validation", UnitID: "unit", Event: entity.IssueEventOpened, BuildID: "build/1", TimestampMs: 1000} - t.Run("history survives resolution and reopening", func(t *testing.T) { - issue := entity.OpenIssue{TaskID: opened.TaskID, UnitID: opened.UnitID, FirstSeenBuildID: opened.BuildID} - require.NoError(t, bound.GetOpenIssueStore().Create(ctx, issue)) + t.Run("history survives closure and reopening", func(t *testing.T) { + issue := entity.IssueState{TaskID: opened.TaskID, UnitID: opened.UnitID, FirstSeenBuildID: opened.BuildID, Status: entity.IssueStatusOpen, Version: 1} + require.NoError(t, bound.GetIssueStateStore().Create(ctx, issue)) require.NoError(t, bound.GetBuildIssueStore().Create(ctx, opened.BuildID, entity.BuildIssue{TaskID: opened.TaskID, UnitID: opened.UnitID})) require.NoError(t, store.Create(ctx, opened)) resolved := opened resolved.Event, resolved.BuildID, resolved.TimestampMs = entity.IssueEventResolved, "build/2", 2000 require.NoError(t, store.Create(ctx, resolved)) - require.NoError(t, bound.GetOpenIssueStore().Delete(ctx, issue)) - _, err := bound.GetOpenIssueStore().Get(ctx, issue.TaskID, issue.UnitID) - require.ErrorIs(t, err, storage.ErrNotFound) + issue.Status = entity.IssueStatusClosed + require.NoError(t, bound.GetIssueStateStore().Update(ctx, issue, 1, 2)) + retained, err := bound.GetIssueStateStore().Get(ctx, issue.TaskID, issue.UnitID) + require.NoError(t, err) + require.Equal(t, entity.IssueStatusClosed, retained.Status) reopened := opened reopened.BuildID, reopened.TimestampMs = "build/3", 3000 issue.FirstSeenBuildID = reopened.BuildID - require.NoError(t, bound.GetOpenIssueStore().Create(ctx, issue)) + issue.Status = entity.IssueStatusOpen + require.NoError(t, bound.GetIssueStateStore().Update(ctx, issue, 2, 3)) require.NoError(t, store.Create(ctx, reopened)) for _, want := range []entity.IssueLog{opened, resolved, reopened} { diff --git a/test/integration/stovepipe/extension/storage/mysql/open_issue_list_test.go b/test/integration/stovepipe/extension/storage/mysql/issue_state_list_test.go similarity index 61% rename from test/integration/stovepipe/extension/storage/mysql/open_issue_list_test.go rename to test/integration/stovepipe/extension/storage/mysql/issue_state_list_test.go index d3f67cf39..b69c94094 100644 --- a/test/integration/stovepipe/extension/storage/mysql/open_issue_list_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/issue_state_list_test.go @@ -26,17 +26,17 @@ import ( "github.com/uber/submitqueue/stovepipe/extension/storage" ) -func testOpenIssueList(t *testing.T, ctx context.Context, factory storage.Factory) { +func testIssueStateList(t *testing.T, ctx context.Context, factory storage.Factory) { t.Helper() - bound, err := factory.For(storage.Config{QueueName: "open-issue-list/main"}) + bound, err := factory.For(storage.Config{QueueName: "issue-state-list/main"}) require.NoError(t, err) - store := bound.GetOpenIssueStore() - empty, err := store.List(ctx, storage.OpenIssueRange{Limit: 2}) + store := bound.GetIssueStateStore() + empty, err := store.List(ctx, storage.IssueStateRange{Limit: 2}) require.NoError(t, err) require.Empty(t, empty) - expected := []entity.OpenIssue{ + expected := []entity.IssueState{ {TaskID: "A", UnitID: ""}, {TaskID: "a", UnitID: ""}, {TaskID: "a", UnitID: "\x00"}, @@ -52,6 +52,11 @@ func testOpenIssueList(t *testing.T, ctx context.Context, factory storage.Factor } for i := range expected { expected[i].FirstSeenBuildID = fmt.Sprintf("build/%d", i) + expected[i].Version = 1 + expected[i].Status = entity.IssueStatusOpen + if i%2 == 0 { + expected[i].Status = entity.IssueStatusClosed + } } for i := len(expected) - 1; i >= 0; i-- { require.NoError(t, store.Create(ctx, expected[i])) @@ -63,16 +68,16 @@ func testOpenIssueList(t *testing.T, ctx context.Context, factory storage.Factor return expected[i].UnitID < expected[j].UnitID }) - other, err := factory.For(storage.Config{QueueName: "Open-issue-list/main"}) + other, err := factory.For(storage.Config{QueueName: "Issue-state-list/main"}) require.NoError(t, err) - require.NoError(t, other.GetOpenIssueStore().Create(ctx, entity.OpenIssue{ - TaskID: "other", UnitID: "unit", FirstSeenBuildID: "other-build", + require.NoError(t, other.GetIssueStateStore().Create(ctx, entity.IssueState{ + TaskID: "other", UnitID: "unit", FirstSeenBuildID: "other-build", Status: entity.IssueStatusOpen, Version: 1, })) for _, limit := range []int{1, 2, 3, len(expected), len(expected) + 1} { t.Run(fmt.Sprintf("page size %d", limit), func(t *testing.T) { - bounds := storage.OpenIssueRange{Limit: limit} - var got []entity.OpenIssue + bounds := storage.IssueStateRange{Limit: limit} + var got []entity.IssueState for { page, err := store.List(ctx, bounds) require.NoError(t, err) @@ -83,7 +88,7 @@ func testOpenIssueList(t *testing.T, ctx context.Context, factory storage.Factor got = append(got, page...) require.LessOrEqual(t, len(got), len(expected)) last := page[len(page)-1] - bounds.After = storage.OpenIssueCursor{TaskID: last.TaskID, UnitID: last.UnitID} + bounds.After = storage.IssueStateCursor{TaskID: last.TaskID, UnitID: last.UnitID} } require.Equal(t, expected, got) }) @@ -91,36 +96,38 @@ func testOpenIssueList(t *testing.T, ctx context.Context, factory storage.Factor t.Run("empty task IDs are rejected", func(t *testing.T) { for _, unitID := range []string{"", "unit"} { - issue := entity.OpenIssue{UnitID: unitID, FirstSeenBuildID: "build/invalid"} + issue := entity.IssueState{UnitID: unitID, FirstSeenBuildID: "build/invalid"} require.Error(t, store.Create(ctx, issue)) _, err := store.Get(ctx, "", unitID) require.Error(t, err) - require.Error(t, store.Delete(ctx, issue)) + require.Error(t, store.Update(ctx, issue, 1, 2)) } - _, err := store.List(ctx, storage.OpenIssueRange{ - After: storage.OpenIssueCursor{UnitID: "unit"}, Limit: 1, + _, err := store.List(ctx, storage.IssueStateRange{ + After: storage.IssueStateCursor{UnitID: "unit"}, Limit: 1, }) require.Error(t, err) }) - t.Run("cursor row need not exist", func(t *testing.T) { - bound, err := factory.For(storage.Config{QueueName: "open-issue-list/deletion"}) + t.Run("closed cursor and absent boundaries remain traversable", func(t *testing.T) { + bound, err := factory.For(storage.Config{QueueName: "issue-state-list/deletion"}) require.NoError(t, err) - store := bound.GetOpenIssueStore() - var expected []entity.OpenIssue + store := bound.GetIssueStateStore() + var expected []entity.IssueState for _, unit := range []string{"a", "b", "c"} { - issue := entity.OpenIssue{TaskID: "task", UnitID: unit, FirstSeenBuildID: "build/1"} + issue := entity.IssueState{TaskID: "task", UnitID: unit, FirstSeenBuildID: "build/1", Status: entity.IssueStatusOpen, Version: 1} expected = append(expected, issue) require.NoError(t, store.Create(ctx, issue)) } - first, err := store.List(ctx, storage.OpenIssueRange{Limit: 1}) + first, err := store.List(ctx, storage.IssueStateRange{Limit: 1}) require.NoError(t, err) require.Equal(t, expected[:1], first) - require.NoError(t, store.Delete(ctx, first[0])) + closed := first[0] + closed.Status = entity.IssueStatusClosed + require.NoError(t, store.Update(ctx, closed, 1, 2)) - bounds := storage.OpenIssueRange{Limit: 2, - After: storage.OpenIssueCursor{TaskID: first[0].TaskID, UnitID: first[0].UnitID}} + bounds := storage.IssueStateRange{Limit: 2, + After: storage.IssueStateCursor{TaskID: first[0].TaskID, UnitID: first[0].UnitID}} rest, err := store.List(ctx, bounds) require.NoError(t, err) require.Equal(t, expected[1:], rest) @@ -134,16 +141,16 @@ func testOpenIssueList(t *testing.T, ctx context.Context, factory storage.Factor t.Run("identifier boundaries", func(t *testing.T) { bound, err := factory.For(storage.Config{QueueName: strings.Repeat("q", 255)}) require.NoError(t, err) - store := bound.GetOpenIssueStore() - issue := entity.OpenIssue{TaskID: strings.Repeat("t", 255), UnitID: strings.Repeat("😀", 256), - FirstSeenBuildID: strings.Repeat("😀", 255)} + store := bound.GetIssueStateStore() + issue := entity.IssueState{TaskID: strings.Repeat("t", 255), UnitID: strings.Repeat("😀", 256), + FirstSeenBuildID: strings.Repeat("😀", 255), Status: entity.IssueStatusOpen, Version: 1} require.NoError(t, store.Create(ctx, issue)) - page, err := store.List(ctx, storage.OpenIssueRange{Limit: 1}) + page, err := store.List(ctx, storage.IssueStateRange{Limit: 1}) require.NoError(t, err) - require.Equal(t, []entity.OpenIssue{issue}, page) + require.Equal(t, []entity.IssueState{issue}, page) - bounds := storage.OpenIssueRange{Limit: 1, - After: storage.OpenIssueCursor{TaskID: issue.TaskID, UnitID: issue.UnitID}} + bounds := storage.IssueStateRange{Limit: 1, + After: storage.IssueStateCursor{TaskID: issue.TaskID, UnitID: issue.UnitID}} page, err = store.List(ctx, bounds) require.NoError(t, err) require.Empty(t, page) diff --git a/test/integration/stovepipe/extension/storage/mysql/issue_state_test.go b/test/integration/stovepipe/extension/storage/mysql/issue_state_test.go new file mode 100644 index 000000000..0f0cbad2b --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/issue_state_test.go @@ -0,0 +1,112 @@ +// 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 mysql + +import ( + "context" + "errors" + "strings" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func testIssueStateStore(t *testing.T, ctx context.Context, factory storage.Factory) { + bound, err := factory.For(storage.Config{QueueName: "issue-state/main"}) + require.NoError(t, err) + store := bound.GetIssueStateStore() + issue := entity.IssueState{TaskID: "validation", UnitID: "unit", Status: entity.IssueStatusOpen, FirstSeenBuildID: "build/1", Version: 1} + require.NoError(t, store.Create(ctx, issue)) + require.ErrorIs(t, store.Create(ctx, issue), storage.ErrAlreadyExists) + got, err := store.Get(ctx, issue.TaskID, issue.UnitID) + require.NoError(t, err) + require.Equal(t, issue, got) + + closed := issue + closed.Status = entity.IssueStatusClosed + require.NoError(t, store.Update(ctx, closed, 1, 2)) + closed.Version = 2 + got, err = store.Get(ctx, issue.TaskID, issue.UnitID) + require.NoError(t, err) + require.Equal(t, closed, got) + require.ErrorIs(t, store.Create(ctx, issue), storage.ErrAlreadyExists) + require.ErrorIs(t, store.Update(ctx, issue, 1, 2), storage.ErrVersionMismatch) + + reopened := closed + reopened.Status = entity.IssueStatusOpen + reopened.FirstSeenBuildID = "build/3" + require.NoError(t, store.Update(ctx, reopened, 2, 3)) + reopened.Version = 3 + require.ErrorIs(t, store.Update(ctx, closed, 2, 3), storage.ErrVersionMismatch) + got, err = store.Get(ctx, issue.TaskID, issue.UnitID) + require.NoError(t, err) + require.Equal(t, reopened, got) + + t.Run("queue and byte-exact key isolation", func(t *testing.T) { + other, err := factory.For(storage.Config{QueueName: "Issue-state/main"}) + require.NoError(t, err) + _, err = other.GetIssueStateStore().Get(ctx, issue.TaskID, issue.UnitID) + require.ErrorIs(t, err, storage.ErrNotFound) + require.NoError(t, other.GetIssueStateStore().Create(ctx, issue)) + for _, key := range []entity.BuildIssue{{TaskID: "validation", UnitID: "Unit"}, {TaskID: "validation", UnitID: "unit "}, {TaskID: "Validation", UnitID: "unit"}, {TaskID: "validation"}, {TaskID: "a\x00b", UnitID: "c"}, {TaskID: "a", UnitID: "b\x00c"}} { + value := issue + value.TaskID, value.UnitID = key.TaskID, key.UnitID + require.NoError(t, store.Create(ctx, value)) + got, err := store.Get(ctx, key.TaskID, key.UnitID) + require.NoError(t, err) + require.Equal(t, value, got) + } + }) + t.Run("only one competing version wins", func(t *testing.T) { + results := make(chan error, 2) + for _, status := range []entity.IssueStatus{entity.IssueStatusOpen, entity.IssueStatusClosed} { + value := reopened + value.Status = status + go func() { results <- store.Update(ctx, value, 3, 4) }() + } + successes, conflicts := 0, 0 + for i := 0; i < 2; i++ { + err := <-results + switch { + case err == nil: + successes++ + case errors.Is(err, storage.ErrVersionMismatch): + conflicts++ + default: + require.NoError(t, err) + } + } + require.Equal(t, 1, successes) + require.Equal(t, 1, conflicts) + }) + t.Run("identifier boundaries", func(t *testing.T) { + bound, err := factory.For(storage.Config{QueueName: strings.Repeat("q", 255)}) + require.NoError(t, err) + value := issue + value.TaskID = strings.Repeat("t", 255) + value.UnitID = strings.Repeat("😀", 256) + value.FirstSeenBuildID = strings.Repeat("😀", 255) + store := bound.GetIssueStateStore() + require.NoError(t, store.Create(ctx, value)) + got, err := store.Get(ctx, value.TaskID, value.UnitID) + require.NoError(t, err) + require.Equal(t, value, got) + value.UnitID += "x" + require.Error(t, store.Create(ctx, value)) + require.Error(t, store.Update(ctx, value, 1, 2)) + }) +} 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 3facd0805..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 @@ -57,10 +57,30 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F got, err = store.Get(ctx) require.NoError(t, err) require.Equal(t, state, got) - opens, err := bound.GetOpenIssueStore().List(ctx, storage.OpenIssueRange{Limit: 1}) + opens, err := bound.GetIssueStateStore().List(ctx, storage.IssueStateRange{Limit: 1}) 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) diff --git a/test/integration/stovepipe/extension/storage/mysql/open_issue_test.go b/test/integration/stovepipe/extension/storage/mysql/open_issue_test.go deleted file mode 100644 index e227e0311..000000000 --- a/test/integration/stovepipe/extension/storage/mysql/open_issue_test.go +++ /dev/null @@ -1,156 +0,0 @@ -// 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 mysql - -import ( - "context" - "errors" - "strings" - "testing" - - "github.com/stretchr/testify/require" - "github.com/uber/submitqueue/stovepipe/entity" - "github.com/uber/submitqueue/stovepipe/extension/storage" -) - -func testOpenIssueStore(t *testing.T, ctx context.Context, factory storage.Factory) { - t.Helper() - - bound, err := factory.For(storage.Config{QueueName: "open-issues/main"}) - require.NoError(t, err) - store := bound.GetOpenIssueStore() - issue := entity.OpenIssue{ - TaskID: "validation", UnitID: "unit", FirstSeenBuildID: "build/1", - } - - require.NoError(t, store.Create(ctx, issue)) - require.ErrorIs(t, store.Create(ctx, issue), storage.ErrAlreadyExists) - got, err := store.Get(ctx, issue.TaskID, issue.UnitID) - require.NoError(t, err) - require.Equal(t, issue, got) - - require.NoError(t, bound.GetBuildIssueStore().Create(ctx, "build/1", entity.BuildIssue{TaskID: issue.TaskID, UnitID: issue.UnitID})) - require.NoError(t, store.Delete(ctx, got)) - require.NoError(t, store.Delete(ctx, got)) - _, err = store.Get(ctx, issue.TaskID, issue.UnitID) - require.ErrorIs(t, err, storage.ErrNotFound) - _, err = bound.GetBuildIssueStore().Get(ctx, "build/1", issue.TaskID, issue.UnitID) - require.NoError(t, err) - - t.Run("stale deletion cannot remove a reopened issue", func(t *testing.T) { - reopened := issue - reopened.FirstSeenBuildID = "build/3" - require.NoError(t, store.Create(ctx, reopened)) - - require.NoError(t, store.Delete(ctx, issue)) - require.NoError(t, store.Delete(ctx, issue)) - require.ErrorIs(t, store.Create(ctx, issue), storage.ErrAlreadyExists) - got, err := store.Get(ctx, issue.TaskID, issue.UnitID) - require.NoError(t, err) - require.Equal(t, reopened, got) - }) - - t.Run("queue and byte-exact key isolation", func(t *testing.T) { - other, err := factory.For(storage.Config{QueueName: "Open-issues/main"}) - require.NoError(t, err) - _, err = other.GetOpenIssueStore().Get(ctx, issue.TaskID, issue.UnitID) - require.ErrorIs(t, err, storage.ErrNotFound) - require.NoError(t, other.GetOpenIssueStore().Delete(ctx, issue)) - require.NoError(t, other.GetOpenIssueStore().Create(ctx, issue)) - require.NoError(t, other.GetOpenIssueStore().Delete(ctx, issue)) - - got, err := store.Get(ctx, issue.TaskID, issue.UnitID) - require.NoError(t, err) - require.Equal(t, "build/3", got.FirstSeenBuildID) - - for _, key := range []entity.BuildIssue{ - {TaskID: "validation", UnitID: "Unit"}, - {TaskID: "validation", UnitID: "unit "}, - {TaskID: "Validation", UnitID: "unit"}, - {TaskID: "validation ", UnitID: "unit"}, - {TaskID: "validation"}, - {TaskID: "a\x00b", UnitID: "c"}, - {TaskID: "a", UnitID: "b\x00c"}, - } { - value := issue - value.TaskID, value.UnitID = key.TaskID, key.UnitID - require.NoError(t, store.Create(ctx, value)) - - got, err := store.Get(ctx, key.TaskID, key.UnitID) - require.NoError(t, err) - require.Equal(t, value, got) - require.NoError(t, store.Delete(ctx, got)) - } - }) - - t.Run("only one concurrent opening wins", func(t *testing.T) { - results := make(chan error, 2) - for _, buildID := range []string{"build/4", "build/5"} { - value := issue - value.UnitID = "concurrent" - value.FirstSeenBuildID = buildID - go func() { results <- store.Create(ctx, value) }() - } - - successes, duplicates := 0, 0 - for i := 0; i < 2; i++ { - err := <-results - switch { - case err == nil: - successes++ - case errors.Is(err, storage.ErrAlreadyExists): - duplicates++ - default: - require.NoError(t, err) - } - } - require.Equal(t, 1, successes) - require.Equal(t, 1, duplicates) - - got, err := store.Get(ctx, issue.TaskID, "concurrent") - require.NoError(t, err) - require.Contains(t, []string{"build/4", "build/5"}, got.FirstSeenBuildID) - }) - - t.Run("binary identifier limits", func(t *testing.T) { - maxBound, err := factory.For(storage.Config{QueueName: strings.Repeat("q", 255)}) - require.NoError(t, err) - maxStore := maxBound.GetOpenIssueStore() - value := entity.OpenIssue{ - TaskID: strings.Repeat("t", 255), UnitID: strings.Repeat("😀", 256), - FirstSeenBuildID: strings.Repeat("😀", 255), - } - require.NoError(t, maxStore.Create(ctx, value)) - - got, err := maxStore.Get(ctx, value.TaskID, value.UnitID) - require.NoError(t, err) - require.Equal(t, value, got) - - oversized := value - oversized.FirstSeenBuildID += "x" - require.Error(t, maxStore.Create(ctx, oversized)) - require.Error(t, maxStore.Delete(ctx, oversized)) - oversized = value - oversized.UnitID += "x" - require.Error(t, maxStore.Create(ctx, oversized)) - require.Error(t, maxStore.Delete(ctx, oversized)) - _, err = maxStore.Get(ctx, oversized.TaskID, oversized.UnitID) - require.Error(t, err) - - got, err = maxStore.Get(ctx, value.TaskID, value.UnitID) - require.NoError(t, err) - require.Equal(t, value, got) - }) -} diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index 380df500d..514bf5b06 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -105,14 +105,14 @@ func TestMySQLStorage(t *testing.T) { testBuildIssueUpdate(t, ctx, db, factory) }) - t.Run("OpenIssueStore", func(t *testing.T) { + t.Run("IssueStateStore", func(t *testing.T) { resetStorage(t, db) - testOpenIssueStore(t, ctx, factory) + testIssueStateStore(t, ctx, factory) }) - t.Run("OpenIssueList", func(t *testing.T) { + t.Run("IssueStateList", func(t *testing.T) { resetStorage(t, db) - testOpenIssueList(t, ctx, factory) + testIssueStateList(t, ctx, factory) }) t.Run("IssueTrackingStore", func(t *testing.T) { @@ -183,7 +183,7 @@ func resetStorage(t *testing.T, db *sql.DB) { t.Helper() for _, statement := range []string{ - "TRUNCATE TABLE open_issue", + "TRUNCATE TABLE issue_state", "TRUNCATE TABLE issue_log", "TRUNCATE TABLE issue_tracking", "TRUNCATE TABLE queue_log",