From 09f2749b1e10924d8e954229bcb8dd97bc17b8ce Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Tue, 27 Oct 2026 00:45:29 +0000 Subject: [PATCH] feat(stovepipe): Apply planned issue changes --- stovepipe/controller/issuetrack/BUILD.bazel | 36 ++ stovepipe/controller/issuetrack/README.md | 13 + stovepipe/controller/issuetrack/issuetrack.go | 117 +++++++ .../controller/issuetrack/issuetrack_test.go | 323 ++++++++++++++++++ .../issuetrack/planned_change_race_test.go | 133 ++++++++ 5 files changed, 622 insertions(+) create mode 100644 stovepipe/controller/issuetrack/BUILD.bazel create mode 100644 stovepipe/controller/issuetrack/README.md create mode 100644 stovepipe/controller/issuetrack/issuetrack.go create mode 100644 stovepipe/controller/issuetrack/issuetrack_test.go create mode 100644 stovepipe/controller/issuetrack/planned_change_race_test.go diff --git a/stovepipe/controller/issuetrack/BUILD.bazel b/stovepipe/controller/issuetrack/BUILD.bazel new file mode 100644 index 000000000..17b348e37 --- /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", + "//platform/metrics: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 = [ + "issuetrack_test.go", + "planned_change_race_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..94cf49f29 --- /dev/null +++ b/stovepipe/controller/issuetrack/README.md @@ -0,0 +1,13 @@ +# Issue tracking + +## Applying a planned change + +The controller applies one `PlannedIssueChange` already saved in the queue's progress record. The cursor identifies the issue; the change holds only its write details. Expected issue version zero means create-if-absent. A positive version means update that exact version. A retry never adopts a newly loaded version to execute old instructions. + +After an issue write succeeds, the controller writes the opening or closing history event, then clears the planned change while retaining the cursor using progress CAS. If an issue write already happened, a retry verifies the exact intended issue state and version before repairing history. The saved timestamp makes duplicate history stable. Failure leaves the original planned change available for another worker to finish. + +A replacement must finish that same planned change before accepting later work. Closed issue rows retain their versions. Together, these rules prevent a paused old worker from reopening or closing an issue after newer work: its create-if-absent or original version guard fails. A conflicting issue state is not treated as a successful replay. + +All writes use complete primary keys. There are no cross-record transactions, leases, snapshot indexes, or additional operation counters. Records and versions must not be deleted or reset while an old delivery could resume. All issue-state writers must follow this protocol. + +This layer does not scan build issues, receive queue messages, publish repairs, or register a runtime consumer. Planning, delivery orchestration, and terminal recovery are separate additions. diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go new file mode 100644 index 000000000..82d41048a --- /dev/null +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -0,0 +1,117 @@ +// 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" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/zap" +) + +// 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 +} + +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, + } +} + +func (c *Controller) applyPlannedChange(ctx context.Context, store storage.Storage, progress entity.IssueTrackingState) (checkpoint entity.IssueTrackingState, retErr error) { + op := metrics.Begin(c.metricsScope, "apply_change", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + + if err := ctx.Err(); err != nil { + return entity.IssueTrackingState{}, err + } + if err := progress.Validate(); err != nil { + return entity.IssueTrackingState{}, err + } + if !progress.HasPlannedChange() { + return entity.IssueTrackingState{}, fmt.Errorf("issue tracking has no saved change") + } + if progress.Version == math.MaxInt32 { + return entity.IssueTrackingState{}, fmt.Errorf("issue tracking version exhausted") + } + + change := progress.PlannedChange + target := entity.IssueState{ + TaskID: progress.CursorTaskID, + UnitID: progress.CursorUnitID, + Status: change.Status, + FirstSeenBuildID: change.FirstSeenBuildID, + Version: change.ExpectedVersion + 1, + } + issues := store.GetIssueStateStore() + var err error + if change.ExpectedVersion == 0 { + err = issues.Create(ctx, target) + } else { + err = issues.Update(ctx, target, change.ExpectedVersion, target.Version) + } + if errors.Is(err, storage.ErrAlreadyExists) || errors.Is(err, storage.ErrVersionMismatch) { + retained, readErr := issues.Get(ctx, progress.CursorTaskID, progress.CursorUnitID) + if readErr != nil { + return entity.IssueTrackingState{}, fmt.Errorf("read planned issue change result: %w", readErr) + } + // A retry may finish history, but must never rebase the planned change onto a newer version. + if retained != target { + return entity.IssueTrackingState{}, fmt.Errorf("planned issue change has been superseded: %w", storage.ErrVersionMismatch) + } + } else if err != nil { + return entity.IssueTrackingState{}, fmt.Errorf("apply planned issue change: %w", err) + } + + event := entity.IssueEventOpened + if change.Status == entity.IssueStatusClosed { + event = entity.IssueEventResolved + } + log := entity.IssueLog{TaskID: progress.CursorTaskID, UnitID: progress.CursorUnitID, BuildID: progress.BuildID, Event: event, TimestampMs: change.TimestampMs} + if err := store.GetIssueLogStore().Create(ctx, log); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + return entity.IssueTrackingState{}, fmt.Errorf("record issue transition: %w", err) + } + + next := progress + next.PlannedChange = entity.PlannedIssueChange{} + newVersion := progress.Version + 1 + if err := store.GetIssueTrackingStore().Update(ctx, next, progress.Version, newVersion); err != nil { + return entity.IssueTrackingState{}, fmt.Errorf("complete planned issue change: %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..d0d857d79 --- /dev/null +++ b/stovepipe/controller/issuetrack/issuetrack_test.go @@ -0,0 +1,323 @@ +// 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" + "sync" + "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 issueKey struct{ task, unit string } +type logKey struct { + issueKey + build string + event entity.IssueEvent +} + +type fixture struct { + mu sync.Mutex + 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 + beforeCreate func() + beforeUpdate func() + beforeHistory func() +} + +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{}} + f.factory = storagemock.NewMockFactory(f.mocks) + 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 + } + f.mu.Lock() + defer f.mu.Unlock() + 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 f.beforeCreate != nil { + f.beforeCreate() + } + if err := f.fail("issue/create"); err != nil { + return err + } + f.mu.Lock() + defer f.mu.Unlock() + 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 f.beforeUpdate != nil { + f.beforeUpdate() + } + if err := f.fail("issue/update"); err != nil { + return err + } + f.mu.Lock() + defer f.mu.Unlock() + 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 f.beforeHistory != nil { + f.beforeHistory() + } + if err := f.fail("history/create"); err != nil { + return err + } + f.mu.Lock() + defer f.mu.Unlock() + key := logKey{issueKey{log.TaskID, log.UnitID}, log.BuildID, log.Event} + 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 + } + f.mu.Lock() + defer f.mu.Unlock() + 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 + } + f.mu.Lock() + defer f.mu.Unlock() + 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 + } + f.mu.Lock() + defer f.mu.Unlock() + if !f.hasProgress || f.progress.Version != oldVersion { + return storage.ErrVersionMismatch + } + state.Version = newVersion + f.progress = state + return nil + }).AnyTimes() + f.controller = NewController(zap.NewNop().Sugar(), tally.NoopScope, f.factory, messagequeue.TopicKeyIssueTrack, "test") + return f +} + +func (f *fixture) fail(name string) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.failures[name] > 0 { + f.failures[name]-- + return assert.AnError + } + return nil +} + +func openingChange() entity.IssueTrackingState { + return entity.IssueTrackingState{RequestID: "100", BuildID: "build/100", NotificationID: "notice/100", CursorTaskID: "task", CursorUnitID: "unit", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1, + PlannedChange: entity.PlannedIssueChange{Status: entity.IssueStatusOpen, FirstSeenBuildID: "build/100", TimestampMs: 1000}} +} + +func (f *fixture) retain(progress entity.IssueTrackingState) { + f.t.Helper() + require.NoError(f.t, progress.Validate()) + f.mu.Lock() + defer f.mu.Unlock() + f.progress = progress + f.hasProgress = true +} + +func (f *fixture) apply(progress entity.IssueTrackingState) (entity.IssueTrackingState, error) { + f.t.Helper() + return f.controller.applyPlannedChange(context.Background(), f.store, progress) +} + +func TestApplyPlannedIssueChanges(t *testing.T) { + for _, name := range []string{"create", "close", "reopen"} { + t.Run(name, func(t *testing.T) { + f := newFixture(t) + progress := openingChange() + switch name { + case "close": + f.states[issueKey{"task", "unit"}] = entity.IssueState{TaskID: "task", UnitID: "unit", Status: entity.IssueStatusOpen, FirstSeenBuildID: "older", Version: 7} + progress.Phase = entity.IssueTrackingPhaseFinalizing + progress.PlannedChange.Status = entity.IssueStatusClosed + progress.PlannedChange.FirstSeenBuildID = "older" + progress.PlannedChange.ExpectedVersion = 7 + case "reopen": + f.states[issueKey{"task", "unit"}] = entity.IssueState{TaskID: "task", UnitID: "unit", Status: entity.IssueStatusClosed, FirstSeenBuildID: "older", Version: 7} + progress.PlannedChange.ExpectedVersion = 7 + } + f.retain(progress) + before := progress + next, err := f.apply(progress) + require.NoError(t, err) + require.Equal(t, before, progress) + require.Equal(t, next, f.progress) + require.False(t, next.HasPlannedChange()) + require.Equal(t, int32(2), next.Version) + require.Equal(t, "task", next.CursorTaskID) + require.Equal(t, "unit", next.CursorUnitID) + issue := f.states[issueKey{"task", "unit"}] + require.Equal(t, progress.PlannedChange.Status, issue.Status) + require.Equal(t, progress.PlannedChange.ExpectedVersion+1, issue.Version) + require.Equal(t, progress.PlannedChange.FirstSeenBuildID, issue.FirstSeenBuildID) + event := entity.IssueEventOpened + if name == "close" { + event = entity.IssueEventResolved + } + require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", event}].TimestampMs) + }) + } +} + +func TestApplyPlannedChangeRetriesPartialWrites(t *testing.T) { + for _, operation := range []string{"issue/create", "issue/update", "history/create", "progress/update"} { + t.Run(operation, func(t *testing.T) { + f := newFixture(t) + progress := openingChange() + if operation == "issue/update" { + progress.PlannedChange.ExpectedVersion = 3 + f.states[issueKey{"task", "unit"}] = entity.IssueState{TaskID: "task", UnitID: "unit", Status: entity.IssueStatusClosed, FirstSeenBuildID: "older", Version: 3} + } + f.retain(progress) + f.failures[operation] = 1 + _, err := f.apply(progress) + require.ErrorIs(t, err, assert.AnError) + require.Equal(t, progress, f.progress) + next, err := f.apply(progress) + require.NoError(t, err) + require.False(t, next.HasPlannedChange()) + require.Len(t, f.logs, 1) + require.Equal(t, progress.PlannedChange.ExpectedVersion+1, f.states[issueKey{"task", "unit"}].Version) + }) + } +} + +func TestApplyPlannedChangeRejectsDifferentStoredResults(t *testing.T) { + for _, name := range []string{"newer version", "different opening", "different status"} { + t.Run(name, func(t *testing.T) { + f := newFixture(t) + progress := openingChange() + f.retain(progress) + existing := entity.IssueState{TaskID: "task", UnitID: "unit", Status: entity.IssueStatusOpen, FirstSeenBuildID: "build/100", Version: 1} + switch name { + case "newer version": + existing.Version = 2 + case "different opening": + existing.FirstSeenBuildID = "other" + case "different status": + existing.Status = entity.IssueStatusClosed + } + f.states[issueKey{"task", "unit"}] = existing + _, err := f.apply(progress) + require.ErrorIs(t, err, storage.ErrVersionMismatch) + require.Equal(t, existing, f.states[issueKey{"task", "unit"}]) + require.Empty(t, f.logs) + require.Equal(t, progress, f.progress) + }) + } +} + +func TestApplyPlannedChangeRejectsMissingOrInvalidWork(t *testing.T) { + f := newFixture(t) + progress := openingChange() + progress.PlannedChange = entity.PlannedIssueChange{} + _, err := f.apply(progress) + require.Error(t, err) + progress = openingChange() + progress.PlannedChange.ExpectedVersion = -1 + _, err = f.apply(progress) + require.Error(t, err) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err = f.controller.applyPlannedChange(ctx, f.store, openingChange()) + require.ErrorIs(t, err, context.Canceled) + require.Empty(t, f.states) + require.Empty(t, f.logs) +} + +func TestPlannedChangeWaitsForReplayVerification(t *testing.T) { + f := newFixture(t) + progress := openingChange() + f.retain(progress) + f.states[issueKey{"task", "unit"}] = entity.IssueState{TaskID: "task", UnitID: "unit", Status: entity.IssueStatusOpen, FirstSeenBuildID: "build/100", Version: 1} + f.failures["issue/get"] = 1 + _, err := f.apply(progress) + require.ErrorIs(t, err, assert.AnError) + require.Equal(t, progress, f.progress) + require.Empty(t, f.logs) + _, err = f.apply(progress) + require.NoError(t, err) + require.Len(t, f.logs, 1) +} diff --git a/stovepipe/controller/issuetrack/planned_change_race_test.go b/stovepipe/controller/issuetrack/planned_change_race_test.go new file mode 100644 index 000000000..01f62288f --- /dev/null +++ b/stovepipe/controller/issuetrack/planned_change_race_test.go @@ -0,0 +1,133 @@ +// 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 ( + "sync/atomic" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +func TestPausedCreateCannotReopenAnIssueAfterNewerClosure(t *testing.T) { + f := newFixture(t) + old := openingChange() + f.retain(old) + paused, resume := make(chan struct{}), make(chan struct{}) + var intercepted atomic.Bool + f.beforeCreate = func() { + if intercepted.CompareAndSwap(false, true) { + close(paused) + <-resume + } + } + result := make(chan error, 1) + go func() { _, err := f.apply(old); result <- err }() + <-paused + + next, err := f.apply(old) + require.NoError(t, err) + closing := next + closing.Version = next.Version + 2 + closing.RequestID = "101" + closing.BuildID = "build/101" + closing.NotificationID = "notice/101" + closing.Phase = entity.IssueTrackingPhaseFinalizing + closing.CursorTaskID, closing.CursorUnitID = "task", "unit" + closing.PlannedChange = entity.PlannedIssueChange{Status: entity.IssueStatusClosed, FirstSeenBuildID: "build/100", ExpectedVersion: 1, TimestampMs: 2000} + f.retain(closing) + _, err = f.apply(closing) + require.NoError(t, err) + + close(resume) + require.ErrorIs(t, <-result, storage.ErrVersionMismatch) + issue := f.states[issueKey{"task", "unit"}] + require.Equal(t, entity.IssueStatusClosed, issue.Status) + require.Equal(t, int32(2), issue.Version) + require.Len(t, f.logs, 2) + require.Equal(t, "build/101", f.progress.BuildID) +} + +func TestPausedCloseCannotOverwriteANewerOpening(t *testing.T) { + f := newFixture(t) + old := openingChange() + old.Phase = entity.IssueTrackingPhaseFinalizing + old.PlannedChange.Status = entity.IssueStatusClosed + old.PlannedChange.FirstSeenBuildID = "older" + old.PlannedChange.ExpectedVersion = 1 + f.states[issueKey{"task", "unit"}] = entity.IssueState{TaskID: "task", UnitID: "unit", Status: entity.IssueStatusOpen, FirstSeenBuildID: "older", Version: 1} + f.retain(old) + paused, resume := make(chan struct{}), make(chan struct{}) + var intercepted atomic.Bool + f.beforeUpdate = func() { + if intercepted.CompareAndSwap(false, true) { + close(paused) + <-resume + } + } + result := make(chan error, 1) + go func() { _, err := f.apply(old); result <- err }() + <-paused + + _, err := f.apply(old) + require.NoError(t, err) + reopening := openingChange() + reopening.RequestID = "101" + reopening.BuildID = "build/101" + reopening.NotificationID = "notice/101" + reopening.Version = f.progress.Version + 2 + reopening.PlannedChange.FirstSeenBuildID = "build/101" + reopening.PlannedChange.ExpectedVersion = 2 + reopening.PlannedChange.TimestampMs = 2000 + f.retain(reopening) + _, err = f.apply(reopening) + require.NoError(t, err) + + close(resume) + require.ErrorIs(t, <-result, storage.ErrVersionMismatch) + issue := f.states[issueKey{"task", "unit"}] + require.Equal(t, entity.IssueStatusOpen, issue.Status) + require.Equal(t, int32(3), issue.Version) + require.Equal(t, "build/101", issue.FirstSeenBuildID) + require.Len(t, f.logs, 2) +} + +func TestLateHistoryWriteCannotDuplicateAnAcceptedTransition(t *testing.T) { + f := newFixture(t) + old := openingChange() + f.retain(old) + paused, resume := make(chan struct{}), make(chan struct{}) + var intercepted atomic.Bool + f.beforeHistory = func() { + if intercepted.CompareAndSwap(false, true) { + close(paused) + <-resume + } + } + result := make(chan error, 1) + go func() { _, err := f.apply(old); result <- err }() + <-paused + + _, err := f.apply(old) + require.NoError(t, err) + close(resume) + require.ErrorIs(t, <-result, storage.ErrVersionMismatch) + require.Len(t, f.logs, 1) + require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventOpened}].TimestampMs) + require.False(t, f.progress.HasPlannedChange()) + require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version) +}