diff --git a/stovepipe/controller/issuetrack/BUILD.bazel b/stovepipe/controller/issuetrack/BUILD.bazel new file mode 100644 index 000000000..441031a49 --- /dev/null +++ b/stovepipe/controller/issuetrack/BUILD.bazel @@ -0,0 +1,36 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["issuetrack.go"], + importpath = "github.com/uber/submitqueue/stovepipe/controller/issuetrack", + visibility = ["//visibility:public"], + deps = [ + "//platform/consumer:go_default_library", + "//stovepipe/core/loader:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = [ + "fixture_test.go", + "issuetrack_test.go", + ], + embed = [":go_default_library"], + deps = [ + "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "//stovepipe/extension/storage/mock:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md new file mode 100644 index 000000000..3af6e37f9 --- /dev/null +++ b/stovepipe/controller/issuetrack/README.md @@ -0,0 +1,19 @@ +# Issue tracking + +Issue tracking maintains current issue state and opening/closing history from persisted build issues. This layer applies one issue at a time; notification handling and bounded passes follow separately. No consumer is registered yet. + +## Applying one issue + +1. Read whether the build reports the issue and load its current state. +2. Reject a source older than the issue's last processed build, using Request order rather than opaque Build IDs. +3. Before changing the issue, conditionally save its task/unit key as the current queue cursor **after** reading the issue row. Repeat this conditional write even when resuming the same cursor. +4. Conditionally create or update the issue row. Record the processed build even when its open/closed status stays the same; update the changing build and timestamp only on an actual transition. +5. Record opening or closing history from the issue row's changing build and timestamp. On retry, the inclusive cursor revisits that row and repairs its history before proceeding. + +Closed issue rows are retained, so their versions cannot reset on reopening. A repeated result from the same build needs no new issue write. Its history is repaired when revisiting the current cursor. + +## Ordering requirements + +The caller must finish the current cursor and its history before moving on or admitting another build. If a replacement worker advances, completing that cursor changes the issue version first. A paused worker's earlier issue write then fails its row guard. A worker that reads later is rejected by source ordering or the conditional cursor write. + +The post-read cursor write and processed-build updates, including unchanged statuses, are required for this ordering. A progress version alone is not a lock around issue writes. These operations use individual K-V reads, conditional writes, and immutable history creates; no cross-record transaction or stored future-change payload is needed. diff --git a/stovepipe/controller/issuetrack/fixture_test.go b/stovepipe/controller/issuetrack/fixture_test.go new file mode 100644 index 000000000..cd24c8f63 --- /dev/null +++ b/stovepipe/controller/issuetrack/fixture_test.go @@ -0,0 +1,289 @@ +// 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 issuetrack + +import ( + "context" + "sort" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +const testQueue = "validation/main" + +type logKey struct { + issueKey + build string + event entity.IssueEvent +} + +type fixture struct { + t *testing.T + mocks *gomock.Controller + controller *Controller + factory *storagemock.MockFactory + store storage.Storage + states map[issueKey]entity.IssueState + logs map[logKey]entity.IssueLog + progress entity.IssueTrackingState + hasProgress bool + failures map[string]int + initialLogs map[logKey]entity.IssueLog + builds map[string]entity.Build + requests map[string]entity.Request + buildIssues map[string]map[issueKey]entity.BuildIssue + sequence int +} + +func newFixture(t *testing.T) *fixture { + t.Helper() + f := &fixture{t: t, mocks: gomock.NewController(t), states: map[issueKey]entity.IssueState{}, logs: map[logKey]entity.IssueLog{}, failures: map[string]int{}, initialLogs: map[logKey]entity.IssueLog{}} + f.factory = storagemock.NewMockFactory(f.mocks) + f.builds = map[string]entity.Build{} + f.requests = map[string]entity.Request{} + f.buildIssues = map[string]map[issueKey]entity.BuildIssue{} + f.source("older", "1", entity.BuildStatusSucceeded) + store := storagemock.NewMockStorage(f.mocks) + f.store = store + states := storagemock.NewMockIssueStateStore(f.mocks) + history := storagemock.NewMockIssueLogStore(f.mocks) + tracking := storagemock.NewMockIssueTrackingStore(f.mocks) + f.factory.EXPECT().For(storage.Config{QueueName: testQueue}).Return(store, nil).AnyTimes() + store.EXPECT().GetIssueStateStore().Return(states).AnyTimes() + store.EXPECT().GetIssueLogStore().Return(history).AnyTimes() + store.EXPECT().GetIssueTrackingStore().Return(tracking).AnyTimes() + + states.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, task, unit string) (entity.IssueState, error) { + if err := f.fail("issue/get"); err != nil { + return entity.IssueState{}, err + } + issue, ok := f.states[issueKey{task, unit}] + if !ok { + return entity.IssueState{}, storage.ErrNotFound + } + return issue, nil + }).AnyTimes() + states.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, issue entity.IssueState) error { + if err := f.fail("issue/create"); err != nil { + return err + } + key := issueKey{issue.TaskID, issue.UnitID} + if _, ok := f.states[key]; ok { + return storage.ErrAlreadyExists + } + f.states[key] = issue + return nil + }).AnyTimes() + states.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, issue entity.IssueState, oldVersion, newVersion int32) error { + if err := f.fail("issue/update"); err != nil { + return err + } + key := issueKey{issue.TaskID, issue.UnitID} + current, ok := f.states[key] + if !ok || current.Version != oldVersion { + return storage.ErrVersionMismatch + } + issue.Version = newVersion + f.states[key] = issue + return nil + }).AnyTimes() + history.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, log entity.IssueLog) error { + if err := f.fail("history/create"); err != nil { + return err + } + key := logKey{issueKey{log.TaskID, log.UnitID}, log.BuildID, log.Event} + if _, ok := f.initialLogs[key]; ok { + return storage.ErrAlreadyExists + } + if _, ok := f.logs[key]; ok { + return storage.ErrAlreadyExists + } + f.logs[key] = log + return nil + }).AnyTimes() + tracking.EXPECT().Get(gomock.Any()).DoAndReturn(func(context.Context) (entity.IssueTrackingState, error) { + if err := f.fail("progress/get"); err != nil { + return entity.IssueTrackingState{}, err + } + if !f.hasProgress { + return entity.IssueTrackingState{}, storage.ErrNotFound + } + return f.progress, nil + }).AnyTimes() + tracking.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, state entity.IssueTrackingState) error { + if err := f.fail("progress/create"); err != nil { + return err + } + if f.hasProgress { + return storage.ErrAlreadyExists + } + f.progress = state + f.hasProgress = true + return nil + }).AnyTimes() + tracking.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, state entity.IssueTrackingState, oldVersion, newVersion int32) error { + if err := f.fail("progress/update"); err != nil { + return err + } + if !f.hasProgress || f.progress.Version != oldVersion { + return storage.ErrVersionMismatch + } + state.Version = newVersion + f.progress = state + return nil + }).AnyTimes() + + builds := storagemock.NewMockBuildStore(f.mocks) + requests := storagemock.NewMockRequestStore(f.mocks) + buildIssues := storagemock.NewMockBuildIssueStore(f.mocks) + store.EXPECT().GetBuildStore().Return(builds).AnyTimes() + store.EXPECT().GetRequestStore().Return(requests).AnyTimes() + store.EXPECT().GetBuildIssueStore().Return(buildIssues).AnyTimes() + builds.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, id string) (entity.Build, error) { + if err := f.fail("build/get"); err != nil { + return entity.Build{}, err + } + value, ok := f.builds[id] + if !ok { + return entity.Build{}, storage.ErrNotFound + } + return value, nil + }).AnyTimes() + requests.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, id string) (entity.Request, error) { + if err := f.fail("request/get"); err != nil { + return entity.Request{}, err + } + value, ok := f.requests[id] + if !ok { + return entity.Request{}, storage.ErrNotFound + } + return value, nil + }).AnyTimes() + buildIssues.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, build, task, unit string) (entity.BuildIssue, error) { + if err := f.fail("build-issues/get"); err != nil { + return entity.BuildIssue{}, err + } + value, ok := f.buildIssues[build][issueKey{task, unit}] + if !ok { + return entity.BuildIssue{}, storage.ErrNotFound + } + return value, nil + }).AnyTimes() + buildIssues.EXPECT().List(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, build string, bounds storage.BuildIssueRange) ([]entity.BuildIssue, error) { + if err := f.fail("build-issues/list"); err != nil { + return nil, err + } + keys := make([]issueKey, 0) + for key := range f.buildIssues[build] { + if key.task > bounds.After.TaskID || (key.task == bounds.After.TaskID && key.unit > bounds.After.UnitID) { + keys = append(keys, key) + } + } + sortKeys(keys) + if len(keys) > bounds.Limit { + keys = keys[:bounds.Limit] + } + result := make([]entity.BuildIssue, 0, len(keys)) + for _, key := range keys { + result = append(result, f.buildIssues[build][key]) + } + return result, nil + }).AnyTimes() + states.EXPECT().List(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, bounds storage.IssueStateRange) ([]entity.IssueState, error) { + if err := f.fail("issue/list"); err != nil { + return nil, err + } + keys := make([]issueKey, 0) + for key := range f.states { + if key.task > bounds.After.TaskID || (key.task == bounds.After.TaskID && key.unit > bounds.After.UnitID) { + keys = append(keys, key) + } + } + sortKeys(keys) + if len(keys) > bounds.Limit { + keys = keys[:bounds.Limit] + } + result := make([]entity.IssueState, 0, len(keys)) + for _, key := range keys { + result = append(result, f.states[key]) + } + return result, nil + }).AnyTimes() + f.controller = NewController(zap.NewNop().Sugar(), tally.NoopScope, f.factory, messagequeue.TopicKeyIssueTrack, "test") + f.controller.nowMs = func() int64 { return 1000 } + return f +} + +func (f *fixture) fail(name string) error { + if f.failures[name] > 0 { + f.failures[name]-- + return assert.AnError + } + return nil +} + +func currentCursor() entity.IssueTrackingState { + return entity.IssueTrackingState{RequestID: "100", BuildID: "build/100", NotificationID: "notice/100", CursorTaskID: "task", CursorUnitID: "unit", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1} +} + +func (f *fixture) retain(progress entity.IssueTrackingState) { + f.t.Helper() + require.NoError(f.t, progress.Validate()) + f.progress, f.hasProgress = progress, true +} + +func sortKeys(keys []issueKey) { + sort.Slice(keys, func(i, j int) bool { + if keys[i].task != keys[j].task { + return keys[i].task < keys[j].task + } + return keys[i].unit < keys[j].unit + }) +} + +func (f *fixture) source(buildID, requestID string, status entity.BuildStatus) { + f.builds[buildID] = entity.Build{ID: buildID, RequestID: requestID, Status: status, Version: 1} + request := entity.Request{ID: requestID, Queue: testQueue, State: entity.RequestStateProcessing, Version: 1} + if status.IsTerminal() { + request.TerminalBuildID = buildID + request.State = entity.RequestStateFailed + } + f.requests[requestID] = request +} + +func (f *fixture) reportIssue(build, task, unit string) { + if f.buildIssues[build] == nil { + f.buildIssues[build] = map[issueKey]entity.BuildIssue{} + } + f.buildIssues[build][issueKey{task, unit}] = entity.BuildIssue{TaskID: task, UnitID: unit} +} + +func (f *fixture) seedIssue(task, unit string, status entity.IssueStatus) { + f.states[issueKey{task, unit}] = entity.IssueState{TaskID: task, UnitID: unit, Status: status, FirstSeenBuildID: "older", Version: 7, LastProcessedBuildID: "older", LastChangedBuildID: "older", ChangedAtMs: 500} + event := entity.IssueEventOpened + if status == entity.IssueStatusClosed { + event = entity.IssueEventResolved + } + f.initialLogs[logKey{issueKey{task, unit}, "older", event}] = entity.IssueLog{TaskID: task, UnitID: unit, BuildID: "older", Event: event, TimestampMs: 500} +} diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go new file mode 100644 index 000000000..72cf7b0b9 --- /dev/null +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -0,0 +1,187 @@ +// 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 issuetrack maintains issue state and lifecycle history from build issues. +package issuetrack + +import ( + "context" + "errors" + "fmt" + "math" + "time" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/stovepipe/core/loader" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/zap" +) + +type issueKey struct{ task, unit string } + +// Controller retains issue transitions and their queue-scoped progress. +type Controller struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + stores storage.Factory + topicKey consumer.TopicKey + consumerGroup string + nowMs func() int64 +} + +func NewController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, topicKey consumer.TopicKey, consumerGroup string) *Controller { + return &Controller{ + logger: logger.Named("issuetrack_controller"), + metricsScope: scope.SubScope("issuetrack_controller"), + stores: stores, + topicKey: topicKey, + consumerGroup: consumerGroup, + nowMs: func() int64 { return time.Now().UnixMilli() }, + } +} + +func (c *Controller) processIssue(ctx context.Context, store storage.Storage, progress entity.IssueTrackingState, key issueKey, reported bool) (entity.IssueTrackingState, error) { + if err := ctx.Err(); err != nil { + return entity.IssueTrackingState{}, err + } + if !reported { + _, err := store.GetBuildIssueStore().Get(ctx, progress.BuildID, key.task, key.unit) + if err != nil && !errors.Is(err, storage.ErrNotFound) { + return entity.IssueTrackingState{}, fmt.Errorf("check reported issue: %w", err) + } + reported = err == nil + } + + if !reported && progress.Phase != entity.IssueTrackingPhaseFinalizing { + return entity.IssueTrackingState{}, fmt.Errorf("refresh cursor requires a reported issue") + } + + desired := entity.IssueStatusClosed + if reported { + desired = entity.IssueStatusOpen + } + + issues := store.GetIssueStateStore() + current, err := issues.Get(ctx, key.task, key.unit) + missing := errors.Is(err, storage.ErrNotFound) + if err != nil && !missing { + return entity.IssueTrackingState{}, fmt.Errorf("read issue state: %w", err) + } + + if missing && !reported { + return entity.IssueTrackingState{}, fmt.Errorf("closing cursor requires an existing issue") + } + + if !missing && current.LastProcessedBuildID != progress.BuildID { + previous, err := loader.ByID(ctx, current.LastProcessedBuildID, store.GetBuildStore().Get, "last processed build") + if err != nil { + return entity.IssueTrackingState{}, err + } + order, err := entity.CompareRequestID(progress.RequestID, previous.RequestID) + if err != nil { + return entity.IssueTrackingState{}, err + } + if order < 0 { + return entity.IssueTrackingState{}, fmt.Errorf("issue already processed a newer source: %w", storage.ErrVersionMismatch) + } + } + + next := progress + next.CursorTaskID, next.CursorUnitID = key.task, key.unit + if !missing && current.LastProcessedBuildID == progress.BuildID && current.Status == desired { + if key == (issueKey{progress.CursorTaskID, progress.CursorUnitID}) { + return next, c.recordHistory(ctx, store, current) + } + return next, nil + } + + // Read the issue before this CAS, including when resuming a cursor. A + // takeover must finish that key before advancing, invalidating this row guard. + next, err = saveProgress(ctx, store.GetIssueTrackingStore(), next) + if err != nil { + return entity.IssueTrackingState{}, err + } + + if !missing { + if current.LastProcessedBuildID == progress.BuildID { + if err := c.recordHistory(ctx, store, current); err != nil { + return entity.IssueTrackingState{}, err + } + } + if current.Version == math.MaxInt32 { + return entity.IssueTrackingState{}, fmt.Errorf("issue state version exhausted") + } + } + + target := current + target.TaskID, target.UnitID = key.task, key.unit + target.LastProcessedBuildID = progress.BuildID + target.Status = desired + target.Version = current.Version + 1 + if missing || current.Status != desired { + target.LastChangedBuildID = progress.BuildID + target.ChangedAtMs = c.nowMs() + if desired == entity.IssueStatusOpen { + target.FirstSeenBuildID = progress.BuildID + } + } + + if missing { + err = issues.Create(ctx, target) + } else { + err = issues.Update(ctx, target, current.Version, target.Version) + } + if errors.Is(err, storage.ErrAlreadyExists) { + return entity.IssueTrackingState{}, fmt.Errorf("issue was concurrently created: %w", storage.ErrVersionMismatch) + } + if err != nil { + return entity.IssueTrackingState{}, fmt.Errorf("apply issue cursor: %w", err) + } + + if missing || current.Status != desired { + return next, c.recordHistory(ctx, store, target) + } + return next, nil +} + +func (c *Controller) recordHistory(ctx context.Context, store storage.Storage, issue entity.IssueState) error { + event := entity.IssueEventOpened + if issue.Status == entity.IssueStatusClosed { + event = entity.IssueEventResolved + } + + log := entity.IssueLog{TaskID: issue.TaskID, UnitID: issue.UnitID, BuildID: issue.LastChangedBuildID, Event: event, TimestampMs: issue.ChangedAtMs} + if err := store.GetIssueLogStore().Create(ctx, log); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + return fmt.Errorf("record issue transition: %w", err) + } + return nil +} + +func saveProgress(ctx context.Context, store storage.IssueTrackingStore, next entity.IssueTrackingState) (entity.IssueTrackingState, error) { + if next.Version == math.MaxInt32 { + return entity.IssueTrackingState{}, fmt.Errorf("issue tracking version exhausted") + } + newVersion := next.Version + 1 + if err := store.Update(ctx, next, next.Version, newVersion); err != nil { + return entity.IssueTrackingState{}, fmt.Errorf("checkpoint issue tracking: %w", err) + } + next.Version = newVersion + return next, nil +} + +func (c *Controller) Name() string { return "issuetrack" } +func (c *Controller) TopicKey() consumer.TopicKey { return c.topicKey } +func (c *Controller) ConsumerGroup() string { return c.consumerGroup } diff --git a/stovepipe/controller/issuetrack/issuetrack_test.go b/stovepipe/controller/issuetrack/issuetrack_test.go new file mode 100644 index 000000000..656c0666b --- /dev/null +++ b/stovepipe/controller/issuetrack/issuetrack_test.go @@ -0,0 +1,134 @@ +// 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 issuetrack + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func TestProcessIssueOpensAndCloses(t *testing.T) { + for _, reported := range []bool{true, false} { + name := "close" + if reported { + name = "open" + } + t.Run(name, func(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusSucceeded) + progress := currentCursor() + progress.Phase = entity.IssueTrackingPhaseFinalizing + progress.CursorTaskID, progress.CursorUnitID = "", "" + f.retain(progress) + if !reported { + f.seedIssue("task", "unit", entity.IssueStatusOpen) + } + + next, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, reported) + require.NoError(t, err) + require.Equal(t, "task", next.CursorTaskID) + require.Equal(t, "unit", next.CursorUnitID) + require.Equal(t, int32(2), next.Version) + issue := f.states[issueKey{"task", "unit"}] + require.Equal(t, "build/100", issue.LastProcessedBuildID) + require.Equal(t, "build/100", issue.LastChangedBuildID) + require.Equal(t, int64(1000), issue.ChangedAtMs) + if reported { + require.Equal(t, entity.IssueStatusOpen, issue.Status) + require.Equal(t, "build/100", issue.FirstSeenBuildID) + require.Equal(t, int32(1), issue.Version) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventOpened}) + } else { + require.Equal(t, entity.IssueStatusClosed, issue.Status) + require.Equal(t, "older", issue.FirstSeenBuildID) + require.Equal(t, int32(8), issue.Version) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventResolved}) + } + }) + } +} + +func TestUnchangedIssueRecordsTheProcessedBuild(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + f.seedIssue("task", "unit", entity.IssueStatusOpen) + progress := currentCursor() + f.retain(progress) + + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) + require.NoError(t, err) + issue := f.states[issueKey{"task", "unit"}] + require.Equal(t, "build/100", issue.LastProcessedBuildID) + require.Equal(t, "older", issue.LastChangedBuildID) + require.Equal(t, "older", issue.FirstSeenBuildID) + require.Equal(t, int64(500), issue.ChangedAtMs) + require.Equal(t, int32(8), issue.Version) + require.Empty(t, f.logs) +} + +func TestHistoryRetryUsesTheStoredTransition(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + f.reportIssue("build/100", "task", "unit") + progress := currentCursor() + f.retain(progress) + f.failures["history/create"] = 1 + + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) + require.ErrorIs(t, err, assert.AnError) + applied := f.states[issueKey{"task", "unit"}] + require.Empty(t, f.logs) + f.controller.nowMs = func() int64 { return 2000 } + + _, err = f.controller.processIssue(context.Background(), f.store, f.progress, issueKey{"task", "unit"}, false) + require.NoError(t, err) + require.Equal(t, applied, f.states[issueKey{"task", "unit"}]) + require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventOpened}].TimestampMs) +} + +func TestCursorFailureDoesNotMutateTheIssue(t *testing.T) { + f := newFixture(t) + progress := currentCursor() + f.retain(progress) + f.failures["progress/update"] = 1 + + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) + require.ErrorIs(t, err, assert.AnError) + require.Empty(t, f.states) + require.Empty(t, f.logs) + require.Equal(t, progress, f.progress) +} + +func TestOlderSourceCannotChangeTheIssue(t *testing.T) { + f := newFixture(t) + f.source("newer", "101", entity.BuildStatusSucceeded) + f.seedIssue("task", "unit", entity.IssueStatusClosed) + current := f.states[issueKey{"task", "unit"}] + current.LastProcessedBuildID = "newer" + f.states[issueKey{"task", "unit"}] = current + progress := currentCursor() + f.retain(progress) + + _, err := f.controller.processIssue(context.Background(), f.store, progress, issueKey{"task", "unit"}, true) + require.ErrorIs(t, err, storage.ErrVersionMismatch) + require.Equal(t, current, f.states[issueKey{"task", "unit"}]) + require.Equal(t, progress, f.progress) + require.Empty(t, f.logs) +}