diff --git a/stovepipe/controller/issuetrack/BUILD.bazel b/stovepipe/controller/issuetrack/BUILD.bazel index 17b348e37..fa9113788 100644 --- a/stovepipe/controller/issuetrack/BUILD.bazel +++ b/stovepipe/controller/issuetrack/BUILD.bazel @@ -6,8 +6,12 @@ go_library( importpath = "github.com/uber/submitqueue/stovepipe/controller/issuetrack", visibility = ["//visibility:public"], deps = [ + "//platform/base/id:go_default_library", + "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//stovepipe/core/loader:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", @@ -18,11 +22,14 @@ go_library( go_test( name = "go_default_test", srcs = [ + "delivery_test.go", "issuetrack_test.go", "planned_change_race_test.go", ], embed = [":go_default_library"], deps = [ + "//platform/base/messagequeue:go_default_library", + "//platform/consumer/mock:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/storage:go_default_library", diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md index 94cf49f29..bd1415599 100644 --- a/stovepipe/controller/issuetrack/README.md +++ b/stovepipe/controller/issuetrack/README.md @@ -1,5 +1,13 @@ # Issue tracking +## Notifications and bounded work + +Notifications identify a queue, build, and whether analysis is final. They share one queue partition and consumer group. A refreshing or finalizing pass is finished before another source is accepted. After a pass completes, request ingest order rejects older builds, finalized requests reject later refreshes, and completed owning notifications are ignored on replay. A notification that helps another pass finish holds once more for its own work. + +Each delivery processes at most 100 distinct task/unit keys, including a resumed change and closed records. Refresh opens newly reported issues but does not close by absence. Finalization also scans issue states: reported issues stay open or reopen, absent open issues close, and closed records remain untouched. A canceled build cannot close by absence because validation may not have run. Every scanned key advances progress, including closed records; a full read requires another batch to establish exhaustion. + +Before changing an issue, planning reads its current version and saves the cursor and intended transition together using progress CAS. No issue write happens if that save loses a race. With a planned change, the cursor identifies the issue to finish; without one, scans continue after the cursor. The planned change retains its original expected version and history timestamp. Final handoff requires successful complete detection and persistence, and the request's recorded winning terminal build. Finalization uses accumulated per-build issues, not just the latest detector response. + ## 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. @@ -10,4 +18,4 @@ A replacement must finish that same planned change before accepting later work. 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. +The reference host does not register this consumer. Repair publication and terminal recovery remain follow-ups. Permanently failing planned changes block later issue-tracking work and require recovery; skipping the planned change is not safe. A future repair handoff must persist its prerequisites and publish before clearing the change that owns that obligation. diff --git a/stovepipe/controller/issuetrack/delivery_test.go b/stovepipe/controller/issuetrack/delivery_test.go new file mode 100644 index 000000000..1bd645630 --- /dev/null +++ b/stovepipe/controller/issuetrack/delivery_test.go @@ -0,0 +1,408 @@ +// 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" + "fmt" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +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} +} + +func (f *fixture) message(build string, final bool) entityqueue.Message { + f.t.Helper() + f.sequence++ + payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{BuildId: build, QueueName: testQueue, Finalize: final}) + require.NoError(f.t, err) + msg := entityqueue.NewMessage(fmt.Sprintf("notice/%d", f.sequence), payload, testQueue, nil) + msg.Tenant = testQueue + return msg +} + +func (f *fixture) deliver(msg entityqueue.Message, hold bool) error { + f.t.Helper() + delivery := consumermock.NewMockDelivery(f.mocks) + delivery.EXPECT().Message().Return(msg) + if hold { + delivery.EXPECT().Hold(int64(0)) + } + return f.controller.Process(context.Background(), delivery) +} + +func TestRefreshOpensReportedIssuesAndKeepsOtherState(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusRunning) + f.seedIssue("old", "", entity.IssueStatusOpen) + f.reportIssue("build", "task", "unit") + msg := f.message("build", false) + require.NoError(t, f.deliver(msg, false)) + require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", ""}].Status) + require.Equal(t, int32(7), f.states[issueKey{"old", ""}].Version) + require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version) + require.Len(t, f.logs, 1) + + f.reportIssue("build", "task", "next") + require.NoError(t, f.deliver(msg, false)) + require.NotContains(t, f.states, issueKey{"task", "next"}) + require.NoError(t, f.deliver(f.message("build", false), false)) + require.Contains(t, f.states, issueKey{"task", "next"}) + require.Len(t, f.logs, 2) + require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version) +} + +func TestFinalizationOpensKeepsAndClosesIssues(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusFailed) + f.seedIssue("a", "", entity.IssueStatusOpen) + f.seedIssue("b", "unit", entity.IssueStatusOpen) + f.seedIssue("closed", "", entity.IssueStatusClosed) + f.reportIssue("build", "b", "unit") + f.reportIssue("build", "c", "unit") + require.NoError(t, f.deliver(f.message("build", true), false)) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"a", ""}].Status) + require.Equal(t, int32(8), f.states[issueKey{"a", ""}].Version) + require.Equal(t, "older", f.states[issueKey{"b", "unit"}].FirstSeenBuildID) + require.Equal(t, int32(7), f.states[issueKey{"b", "unit"}].Version) + require.Equal(t, "build", f.states[issueKey{"c", "unit"}].FirstSeenBuildID) + require.Equal(t, int32(7), f.states[issueKey{"closed", ""}].Version) + require.Len(t, f.logs, 2) + require.Contains(t, f.logs, logKey{issueKey{"a", ""}, "build", entity.IssueEventResolved}) + require.Contains(t, f.logs, logKey{issueKey{"c", "unit"}, "build", entity.IssueEventOpened}) +} + +func TestEmptyFinalBuildClosesIssuesButRetainsTheirRows(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusSucceeded) + f.seedIssue("task", "", entity.IssueStatusOpen) + require.NoError(t, f.deliver(f.message("build", true), false)) + require.Len(t, f.states, 1) + require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", ""}].Status) + require.Len(t, f.logs, 1) +} + +func TestReportedIssueReopensTheSameRecord(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusRunning) + f.seedIssue("task", "unit", entity.IssueStatusClosed) + f.reportIssue("build", "task", "unit") + require.NoError(t, f.deliver(f.message("build", false), false)) + issue := f.states[issueKey{"task", "unit"}] + require.Equal(t, entity.IssueStatusOpen, issue.Status) + require.Equal(t, int32(8), issue.Version) + require.Equal(t, "build", issue.FirstSeenBuildID) +} + +func TestClosedRecordsStillAdvanceTheCursor(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusSucceeded) + for i := 0; i < 101; i++ { + f.seedIssue("task", fmt.Sprintf("unit/%03d", i), entity.IssueStatusClosed) + } + msg := f.message("build", true) + require.NoError(t, f.deliver(msg, true)) + require.Equal(t, "unit/099", f.progress.CursorUnitID) + require.Equal(t, entity.IssueTrackingPhaseFinalizing, f.progress.Phase) + require.False(t, f.progress.HasPlannedChange()) + require.Empty(t, f.logs) + require.NoError(t, f.deliver(msg, false)) + require.Equal(t, "unit/100", f.progress.CursorUnitID) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + require.Empty(t, f.logs) +} + +func TestRefreshCheckpointsAtTheBatchBoundary(t *testing.T) { + for _, count := range []int{0, 99, 100, 103} { + t.Run(fmt.Sprint(count), func(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusRunning) + for i := 0; i < count; i++ { + f.reportIssue("build", "task", fmt.Sprintf("unit/%03d", i)) + } + msg := f.message("build", false) + require.NoError(t, f.deliver(msg, count >= 100)) + require.Len(t, f.states, min(count, 100)) + if count >= 100 { + require.Equal(t, "unit/099", f.progress.CursorUnitID) + require.NoError(t, f.deliver(msg, false)) + } + require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) + require.Len(t, f.states, count) + require.Len(t, f.logs, count) + }) + } +} + +func TestShortReadsWithMoreThanOneBatchOfCombinedKeys(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusFailed) + for i := 0; i < 99; i++ { + f.reportIssue("build", "task", fmt.Sprintf("unit/%03d", 2*i)) + f.seedIssue("task", fmt.Sprintf("unit/%03d", 2*i+1), entity.IssueStatusOpen) + } + msg := f.message("build", true) + require.NoError(t, f.deliver(msg, true)) + require.Equal(t, "unit/099", f.progress.CursorUnitID) + require.NoError(t, f.deliver(msg, false)) + require.Equal(t, "unit/197", f.progress.CursorUnitID) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + open, closed := 0, 0 + for _, issue := range f.states { + if issue.Status == entity.IssueStatusOpen { + open++ + } else { + closed++ + } + } + require.Equal(t, 99, open) + require.Equal(t, 99, closed) + require.Len(t, f.logs, 198) +} + +func TestCanceledBuildKeepsExistingIssuesOpen(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusCancelled) + f.seedIssue("old", "", entity.IssueStatusOpen) + f.reportIssue("build", "new", "") + f.failures["issue/list"] = 1 + require.NoError(t, f.deliver(f.message("build", true), false)) + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", ""}].Status) + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"new", ""}].Status) + require.Len(t, f.logs, 1) +} + +func TestFinalNotificationBeforeLateRefreshAndOlderBuild(t *testing.T) { + f := newFixture(t) + f.source("older-build", "9", entity.BuildStatusSucceeded) + f.source("newer-build", "10", entity.BuildStatusSucceeded) + require.NoError(t, f.deliver(f.message("newer-build", true), false)) + before := f.progress + f.reportIssue("newer-build", "late", "") + f.reportIssue("older-build", "old", "") + require.NoError(t, f.deliver(f.message("newer-build", false), false)) + require.NoError(t, f.deliver(f.message("older-build", true), false)) + require.Equal(t, before, f.progress) + require.Empty(t, f.states) +} + +func TestAnotherNotificationFinishesSavedWorkBeforeItsOwnBuild(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + f.source("build/101", "101", entity.BuildStatusSucceeded) + f.reportIssue("build/100", "task", "unit") + f.retain(openingChange()) + incoming := f.message("build/101", true) + require.NoError(t, f.deliver(incoming, true)) + require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) + require.Equal(t, "build/100", f.progress.BuildID) + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"task", "unit"}].Status) + require.NoError(t, f.deliver(incoming, false)) + require.Equal(t, "build/101", f.progress.BuildID) + require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", "unit"}].Status) + require.Equal(t, int32(2), f.states[issueKey{"task", "unit"}].Version) +} + +func TestIssueMutationWaitsForItsPlannedChange(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusRunning) + f.reportIssue("build", "task", "unit") + f.failures["progress/update"] = 1 + msg := f.message("build", false) + require.ErrorIs(t, f.deliver(msg, false), assert.AnError) + require.Empty(t, f.states) + require.Empty(t, f.logs) + require.False(t, f.progress.HasPlannedChange()) + require.NoError(t, f.deliver(msg, false)) + require.Len(t, f.states, 1) + require.Len(t, f.logs, 1) +} + +func TestRetryRepairsHistoryBeforeFinalizing(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusFailed) + f.reportIssue("build", "task", "unit") + f.failures["history/create"] = 1 + msg := f.message("build", true) + require.ErrorIs(t, f.deliver(msg, false), assert.AnError) + saved := f.progress.PlannedChange + require.True(t, f.progress.HasPlannedChange()) + require.Equal(t, "task", f.progress.CursorTaskID) + require.Equal(t, "unit", f.progress.CursorUnitID) + require.Equal(t, int32(0), saved.ExpectedVersion) + require.NoError(t, f.deliver(msg, false)) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + require.False(t, f.progress.HasPlannedChange()) + require.Equal(t, "task", f.progress.CursorTaskID) + require.Equal(t, "unit", f.progress.CursorUnitID) + require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version) + require.Equal(t, saved.TimestampMs, f.logs[logKey{issueKey{"task", "unit"}, "build", entity.IssueEventOpened}].TimestampMs) +} + +func TestRetryFinishesTheCursorIssueBeforeScanningTheNextBatch(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusRunning) + for i := 0; i < 101; i++ { + f.reportIssue("build", "task", fmt.Sprintf("unit/%03d", i)) + } + f.failures["history/create"] = 1 + msg := f.message("build", false) + + require.ErrorIs(t, f.deliver(msg, false), assert.AnError) + require.Equal(t, "unit/000", f.progress.CursorUnitID) + require.True(t, f.progress.HasPlannedChange()) + require.Len(t, f.states, 1) + require.Empty(t, f.logs) + + require.NoError(t, f.deliver(msg, true)) + require.Equal(t, "unit/099", f.progress.CursorUnitID) + require.False(t, f.progress.HasPlannedChange()) + require.Len(t, f.states, 100) + require.Len(t, f.logs, 100) + require.Equal(t, int32(1), f.states[issueKey{"task", "unit/000"}].Version) + + require.NoError(t, f.deliver(msg, false)) + require.Equal(t, "unit/100", f.progress.CursorUnitID) + require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) + require.Len(t, f.states, 101) + require.Len(t, f.logs, 101) +} + +func TestPlanningFailuresDoNotChangeIssues(t *testing.T) { + for _, operation := range []string{"build-issues/list", "issue/list", "build-issues/get", "issue/get"} { + t.Run(operation, func(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusSucceeded) + f.seedIssue("task", "unit", entity.IssueStatusOpen) + f.failures[operation] = 1 + require.ErrorIs(t, f.deliver(f.message("build", true), false), assert.AnError) + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"task", "unit"}].Status) + require.Empty(t, f.logs) + require.False(t, f.progress.HasPlannedChange()) + }) + } +} + +func TestFinalizationRequiresTheWinningTerminalBuild(t *testing.T) { + for _, status := range []entity.BuildStatus{entity.BuildStatusRunning, entity.BuildStatusSucceeded} { + t.Run(string(status), func(t *testing.T) { + f := newFixture(t) + f.source("build", "10", status) + request := f.requests["10"] + request.TerminalBuildID = "" + f.requests["10"] = request + require.Error(t, f.deliver(f.message("build", true), false)) + require.False(t, f.hasProgress) + require.Empty(t, f.states) + }) + } + f := newFixture(t) + f.source("loser", "10", entity.BuildStatusFailed) + request := f.requests["10"] + request.TerminalBuildID = "winner" + f.requests["10"] = request + require.NoError(t, f.deliver(f.message("loser", true), false)) + require.False(t, f.hasProgress) +} + +func TestProcessRejectsInvalidMessageAndSource(t *testing.T) { + for _, kind := range []string{"payload", "partition", "tenant", "missing message ID", "missing build", "request queue", "invalid request ID"} { + t.Run(kind, func(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusRunning) + msg := f.message("build", false) + switch kind { + case "payload": + msg.Payload = []byte("invalid") + case "partition": + msg.PartitionKey = "other" + case "tenant": + msg.Tenant = "other" + case "missing message ID": + msg.ID = "" + case "missing build": + delete(f.builds, "build") + case "request queue": + request := f.requests["10"] + request.Queue = "other" + f.requests["10"] = request + case "invalid request ID": + request := f.requests["10"] + request.ID = "010" + f.requests["10"] = request + } + require.Error(t, f.deliver(msg, false)) + require.False(t, f.hasProgress) + require.Empty(t, f.states) + }) + } +} + +func TestLosingThePlannedChangeCASCannotWriteAnIssue(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + f.source("build/101", "101", entity.BuildStatusSucceeded) + f.reportIssue("build/100", "task", "unit") + old := f.message("build/100", false) + paused, resume := make(chan struct{}), make(chan struct{}) + var intercepted atomic.Bool + f.beforeProgressUpdate = func(next entity.IssueTrackingState) { + if next.HasPlannedChange() && intercepted.CompareAndSwap(false, true) { + close(paused) + <-resume + } + } + result := make(chan error, 1) + go func() { result <- f.deliver(old, false) }() + <-paused + + require.NoError(t, f.deliver(old, false)) + require.NoError(t, f.deliver(f.message("build/101", true), false)) + require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", "unit"}].Status) + close(resume) + require.ErrorIs(t, <-result, storage.ErrVersionMismatch) + require.Equal(t, int32(2), f.states[issueKey{"task", "unit"}].Version) + require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", "unit"}].Status) + require.Len(t, f.logs, 2) +} diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go index 82d41048a..d9d3414c0 100644 --- a/stovepipe/controller/issuetrack/issuetrack.go +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -20,15 +20,26 @@ import ( "errors" "fmt" "math" + "sort" + "time" "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/base/id" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/core/loader" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/zap" ) +// Each delivery scans at most this many distinct task/unit keys, including closed records. +const _issueBatchSize = 100 + +type issueKey struct{ task, unit string } + // Controller retains issue transitions and their queue-scoped progress. type Controller struct { logger *zap.SugaredLogger @@ -36,6 +47,7 @@ type Controller struct { 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 { @@ -45,7 +57,268 @@ func NewController(logger *zap.SugaredLogger, scope tally.Scope, stores storage. stores: stores, topicKey: topicKey, consumerGroup: consumerGroup, + nowMs: func() int64 { return time.Now().UnixMilli() }, + } +} + +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { + msg := delivery.Message() + notification := &stovepipemq.IssueTrack{} + if err := stovepipemq.Unmarshal(msg.Payload, notification); err != nil { + return fmt.Errorf("deserialize issue tracking notification: %w", err) + } + queue := notification.GetQueueName() + if err := entityqueue.ValidatePayloadQueue(msg, queue); err != nil { + return fmt.Errorf("invalid message identity: %w", err) + } + if msg.ID == "" || queue == "" || notification.GetBuildId() == "" { + return fmt.Errorf("issue tracking requires message, queue, and build identifiers") + } + if msg.PartitionKey != queue { + return fmt.Errorf("issue tracking must be partitioned by queue %q", queue) + } + store, err := c.stores.For(storage.Config{QueueName: queue}) + if err != nil { + return fmt.Errorf("resolve issue tracking storage: %w", err) + } + return c.trackBuild(ctx, delivery, store, notification, msg.ID) +} + +func (c *Controller) trackBuild(ctx context.Context, delivery consumer.Delivery, store storage.Storage, notification *stovepipemq.IssueTrack, notificationID string) error { + queue := notification.GetQueueName() + build, request, err := loadBuildAndRequest(ctx, store, queue, notification.GetBuildId()) + if err != nil { + return err + } + + tracking := store.GetIssueTrackingStore() + for { + if err := ctx.Err(); err != nil { + return err + } + progress, err := tracking.Get(ctx) + if errors.Is(err, storage.ErrNotFound) { + progress = entity.IssueTrackingState{} + } else if err != nil { + return fmt.Errorf("load issue tracking progress: %w", err) + } + + if progress.Phase != entity.IssueTrackingPhaseRefreshing && progress.Phase != entity.IssueTrackingPhaseFinalizing { + if request.TerminalBuildID != "" && request.TerminalBuildID != build.ID { + return nil + } + if progress.Version != 0 { + if progress.NotificationID == notificationID { + return nil + } + order, err := entity.CompareRequestID(request.ID, progress.RequestID) + if err != nil { + return err + } + if order < 0 || (order == 0 && progress.Phase == entity.IssueTrackingPhaseFinalized) { + c.logger.Debugw("ignoring superseded issue tracking notification", + "queue", queue, "build_id", build.ID, "request_id", request.ID, "current_request_id", progress.RequestID) + return nil + } + } + + if progress.Version == math.MaxInt32 { + return fmt.Errorf("issue tracking version exhausted") + } + next := entity.IssueTrackingState{ + RequestID: request.ID, + BuildID: build.ID, + NotificationID: notificationID, + Phase: entity.IssueTrackingPhaseRefreshing, + Version: progress.Version + 1, + } + if notification.GetFinalize() { + if !build.Status.IsTerminal() || request.TerminalBuildID != build.ID { + return fmt.Errorf("final issue tracking requires the request's terminal build") + } + next.Phase = entity.IssueTrackingPhaseFinalizing + } + + if progress.Version == 0 { + err = tracking.Create(ctx, next) + } else { + err = tracking.Update(ctx, next, progress.Version, next.Version) + } + if errors.Is(err, storage.ErrAlreadyExists) || errors.Is(err, storage.ErrVersionMismatch) { + continue + } + if err != nil { + return fmt.Errorf("start issue tracking pass: %w", err) + } + progress = next + } + + checkpoint, err := c.reconcileBatch(ctx, store, queue, progress) + if err != nil { + return err + } + if checkpoint.NotificationID != notificationID || + checkpoint.Phase == entity.IssueTrackingPhaseRefreshing || + checkpoint.Phase == entity.IssueTrackingPhaseFinalizing { + delivery.Hold(0) + } + return nil + } +} + +func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage, queue string, progress entity.IssueTrackingState) (checkpoint entity.IssueTrackingState, retErr error) { + op := metrics.Begin(c.metricsScope, "batch", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...) + defer func() { op.Complete(retErr) }() + + limit := _issueBatchSize + if progress.HasPlannedChange() { + next, err := c.applyPlannedChange(ctx, store, progress) + if err != nil { + return entity.IssueTrackingState{}, err + } + progress = next + limit-- + } + + build, request, err := loadBuildAndRequest(ctx, store, queue, progress.BuildID) + if err != nil { + return entity.IssueTrackingState{}, err + } + switch progress.Phase { + case entity.IssueTrackingPhaseRefreshing, entity.IssueTrackingPhaseFinalizing: + default: + return entity.IssueTrackingState{}, fmt.Errorf("issue batch requires a refreshing or finalizing pass") + } + if request.ID != progress.RequestID { + return entity.IssueTrackingState{}, fmt.Errorf("issue batch does not match its source request") + } + finalizing := progress.Phase == entity.IssueTrackingPhaseFinalizing + if finalizing && (!build.Status.IsTerminal() || request.TerminalBuildID != build.ID) { + return entity.IssueTrackingState{}, fmt.Errorf("final issue tracking requires the request's terminal build") + } + + buildIssues, err := store.GetBuildIssueStore().List(ctx, build.ID, storage.BuildIssueRange{ + After: storage.BuildIssueCursor{TaskID: progress.CursorTaskID, UnitID: progress.CursorUnitID}, + Limit: limit, + }) + if err != nil { + return entity.IssueTrackingState{}, fmt.Errorf("list build issues: %w", err) + } + + var states []entity.IssueState + // Cancellation may have skipped validation, so absence cannot close an existing issue. + if finalizing && build.Status != entity.BuildStatusCancelled { + states, err = store.GetIssueStateStore().List(ctx, storage.IssueStateRange{ + After: storage.IssueStateCursor{TaskID: progress.CursorTaskID, UnitID: progress.CursorUnitID}, + Limit: limit, + }) + if err != nil { + return entity.IssueTrackingState{}, fmt.Errorf("list issue states: %w", err) + } + } + + reported := make(map[issueKey]bool, len(buildIssues)+len(states)) + for _, issue := range states { + reported[issueKey{issue.TaskID, issue.UnitID}] = false + } + for _, issue := range buildIssues { + reported[issueKey{issue.TaskID, issue.UnitID}] = true } + keys := make([]issueKey, 0, len(reported)) + for key := range reported { + keys = append(keys, key) + } + 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 + }) + complete := len(buildIssues) < limit && len(states) < limit && len(keys) <= limit + if len(keys) > limit { + keys = keys[:limit] + } + + cursorChanged := false + for _, key := range keys { + if err := ctx.Err(); err != nil { + return entity.IssueTrackingState{}, err + } + change, err := c.planIssueChange(ctx, store, progress, key, reported[key]) + if err != nil { + return entity.IssueTrackingState{}, err + } + if change == (entity.PlannedIssueChange{}) { + progress.CursorTaskID, progress.CursorUnitID = key.task, key.unit + cursorChanged = true + continue + } + + next := progress + next.CursorTaskID, next.CursorUnitID = key.task, key.unit + next.PlannedChange = change + progress, err = saveProgress(ctx, store.GetIssueTrackingStore(), next) + if err != nil { + return entity.IssueTrackingState{}, err + } + progress, err = c.applyPlannedChange(ctx, store, progress) + if err != nil { + return entity.IssueTrackingState{}, err + } + cursorChanged = false + } + + if complete { + next := progress + next.Phase = entity.IssueTrackingPhaseRefreshed + if finalizing { + next.Phase = entity.IssueTrackingPhaseFinalized + } + return saveProgress(ctx, store.GetIssueTrackingStore(), next) + } + if cursorChanged { + return saveProgress(ctx, store.GetIssueTrackingStore(), progress) + } + + return progress, nil +} + +func (c *Controller) planIssueChange(ctx context.Context, store storage.Storage, progress entity.IssueTrackingState, key issueKey, reported bool) (entity.PlannedIssueChange, error) { + if !reported { + _, err := store.GetBuildIssueStore().Get(ctx, progress.BuildID, key.task, key.unit) + if err != nil && !errors.Is(err, storage.ErrNotFound) { + return entity.PlannedIssueChange{}, fmt.Errorf("check build issue before closing: %w", err) + } + reported = err == nil + } + + desired := entity.IssueStatusClosed + if reported { + desired = entity.IssueStatusOpen + } + + current, err := store.GetIssueStateStore().Get(ctx, key.task, key.unit) + if err != nil && !errors.Is(err, storage.ErrNotFound) { + return entity.PlannedIssueChange{}, fmt.Errorf("read issue state: %w", err) + } + if (errors.Is(err, storage.ErrNotFound) && desired == entity.IssueStatusClosed) || current.Status == desired { + return entity.PlannedIssueChange{}, nil + } + + firstSeen := current.FirstSeenBuildID + if desired == entity.IssueStatusOpen { + firstSeen = progress.BuildID + } + change := entity.PlannedIssueChange{ + Status: desired, + FirstSeenBuildID: firstSeen, + ExpectedVersion: current.Version, + TimestampMs: c.nowMs(), + } + if err := change.Validate(); err != nil { + return entity.PlannedIssueChange{}, err + } + return change, nil } func (c *Controller) applyPlannedChange(ctx context.Context, store storage.Storage, progress entity.IssueTrackingState) (checkpoint entity.IssueTrackingState, retErr error) { @@ -112,6 +385,38 @@ func (c *Controller) applyPlannedChange(ctx context.Context, store storage.Stora return next, nil } +func saveProgress(ctx context.Context, store storage.IssueTrackingStore, next entity.IssueTrackingState) (entity.IssueTrackingState, error) { + if next.Version == math.MaxInt32 || (next.HasPlannedChange() && next.Version == math.MaxInt32-1) { + 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 loadBuildAndRequest(ctx context.Context, store storage.Storage, queue, buildID string) (entity.Build, entity.Request, error) { + build, err := loader.ByID(ctx, buildID, store.GetBuildStore().Get, "build") + if err != nil { + return entity.Build{}, entity.Request{}, err + } + request, err := loader.ByID(ctx, build.RequestID, store.GetRequestStore().Get, "request") + if err != nil { + return entity.Build{}, entity.Request{}, err + } + if request.Queue != queue { + return entity.Build{}, entity.Request{}, fmt.Errorf("issue tracking request belongs to another queue") + } + if err := id.Validate(request.ID); err != nil { + return entity.Build{}, entity.Request{}, fmt.Errorf("invalid issue tracking request ID: %w", err) + } + return build, request, 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 } + +var _ consumer.Controller = (*Controller)(nil) diff --git a/stovepipe/controller/issuetrack/issuetrack_test.go b/stovepipe/controller/issuetrack/issuetrack_test.go index d0d857d79..8fa0a99d6 100644 --- a/stovepipe/controller/issuetrack/issuetrack_test.go +++ b/stovepipe/controller/issuetrack/issuetrack_test.go @@ -16,6 +16,7 @@ package issuetrack import ( "context" + "sort" "sync" "testing" @@ -32,7 +33,6 @@ import ( const testQueue = "validation/main" -type issueKey struct{ task, unit string } type logKey struct { issueKey build string @@ -40,26 +40,34 @@ type logKey struct { } 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() + 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() + beforeProgressUpdate func(entity.IssueTrackingState) + 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{}} 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{} store := storagemock.NewMockStorage(f.mocks) f.store = store states := storagemock.NewMockIssueStateStore(f.mocks) @@ -157,6 +165,9 @@ func newFixture(t *testing.T) *fixture { 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 f.beforeProgressUpdate != nil { + f.beforeProgressUpdate(state) + } if err := f.fail("progress/update"); err != nil { return err } @@ -169,7 +180,95 @@ func newFixture(t *testing.T) *fixture { 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 + } + f.mu.Lock() + defer f.mu.Unlock() + 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 + } + f.mu.Lock() + defer f.mu.Unlock() + 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 + } + f.mu.Lock() + defer f.mu.Unlock() + 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 + } + f.mu.Lock() + defer f.mu.Unlock() + 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 + } + f.mu.Lock() + defer f.mu.Unlock() + 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 } @@ -321,3 +420,12 @@ func TestPlannedChangeWaitsForReplayVerification(t *testing.T) { require.NoError(t, err) require.Len(t, f.logs, 1) } + +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 + }) +} diff --git a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel index e4a69b5aa..93a9241aa 100644 --- a/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/test/integration/stovepipe/extension/storage/mysql/BUILD.bazel @@ -9,6 +9,7 @@ go_test( "issue_log_test.go", "issue_state_list_test.go", "issue_state_test.go", + "issue_tracking_delivery_test.go", "issue_tracking_test.go", "project_fact_by_degree_test.go", "queue_state_transitions_test.go", @@ -18,6 +19,7 @@ go_test( ], data = [ "docker-compose.yml", + "//platform/extension/messagequeue/mysql/schema", "//stovepipe/extension/storage/mysql/schema", ], tags = [ @@ -25,6 +27,14 @@ go_test( "requires-network", ], deps = [ + "//platform/consumer:go_default_library", + "//platform/errs:go_default_library", + "//platform/extension/consumergate/noop:go_default_library", + "//platform/extension/messagequeue:go_default_library", + "//platform/extension/messagequeue/mysql:go_default_library", + "//platform/publish:go_default_library", + "//stovepipe/controller/issuetrack:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/core/queuelog:go_default_library", "//stovepipe/core/queuestate:go_default_library", "//stovepipe/core/requestlog:go_default_library", @@ -40,5 +50,6 @@ go_test( "@com_github_stretchr_testify//require:go_default_library", "@com_github_stretchr_testify//suite:go_default_library", "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//:go_default_library", ], ) diff --git a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_delivery_test.go b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_delivery_test.go new file mode 100644 index 000000000..b63430ff4 --- /dev/null +++ b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_delivery_test.go @@ -0,0 +1,271 @@ +// 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" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/errs" + gatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop" + extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" + queuemysql "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" + "github.com/uber/submitqueue/platform/publish" + "github.com/uber/submitqueue/stovepipe/controller/issuetrack" + mq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "github.com/uber/submitqueue/test/testutil" + "go.uber.org/zap" +) + +type trackingDeliveryResult struct { + id string + held bool + err error + progress entity.IssueTrackingState +} + +type observedTrackingDelivery struct { + consumer.Delivery + held bool +} + +func (d *observedTrackingDelivery) Hold(delayMs int64) { + d.held = true + d.Delivery.Hold(delayMs) +} + +type trackingObserver struct { + consumer.Controller + progress storage.IssueTrackingStore + results chan trackingDeliveryResult +} + +func (o *trackingObserver) Process(ctx context.Context, delivery consumer.Delivery) error { + observed := &observedTrackingDelivery{Delivery: delivery} + err := o.Controller.Process(ctx, observed) + progress, readErr := o.progress.Get(ctx) + resultErr := err + if resultErr == nil { + resultErr = readErr + } + o.results <- trackingDeliveryResult{delivery.Message().ID, observed.held, resultErr, progress} + return err +} + +func testIssueTrackingDelivery(t *testing.T, ctx context.Context, db *sql.DB, factory storage.Factory) { + t.Helper() + const queueName = "tracking/delivery" + const buildID = "build/delivery" + bound, err := factory.For(storage.Config{QueueName: queueName}) + require.NoError(t, err) + request := entity.Request{ID: "10", Queue: queueName, URI: "git://repo/tracking/delivery", State: entity.RequestStateProcessing, Version: 1} + require.NoError(t, bound.GetRequestStore().Create(ctx, request)) + request.State, request.TerminalBuildID = entity.RequestStateFailed, buildID + require.NoError(t, bound.GetRequestStore().Update(ctx, request, 1, 2)) + require.NoError(t, bound.GetBuildStore().Create(ctx, entity.Build{ID: buildID, RequestID: request.ID, Status: entity.BuildStatusFailed, Version: 1})) + for i := 0; i < 103; i++ { + require.NoError(t, bound.GetBuildIssueStore().Create(ctx, buildID, entity.BuildIssue{TaskID: "task", UnitID: fmt.Sprintf("unit/%03d", i)})) + } + require.NoError(t, bound.GetIssueStateStore().Create(ctx, entity.IssueState{TaskID: "z", FirstSeenBuildID: "old", Status: entity.IssueStatusOpen, Version: 1})) + + testutil.ApplySchema(t, testutil.NewTestLogger(t), db, testutil.SchemaDir("platform/extension/messagequeue/mysql/schema")) + queue, err := queuemysql.NewQueue(queuemysql.Params{DB: db, Logger: zap.NewNop(), MetricsScope: tally.NoopScope, Tenants: []string{queueName}}) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, queue.Close()) }) + config := extqueue.DefaultSubscriptionConfig("issue-tracking-test", "issue-tracking-test") + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{{Key: mq.TopicKeyIssueTrack, Name: "issue_tracking_test", Queue: queue, Subscription: config}}) + require.NoError(t, err) + results := make(chan trackingDeliveryResult, 64) + controller := issuetrack.NewController(zap.NewNop().Sugar(), tally.NoopScope, factory, mq.TopicKeyIssueTrack, config.ConsumerGroup) + observer := &trackingObserver{Controller: controller, progress: bound.GetIssueTrackingStore(), results: results} + runtime := consumer.New(zap.NewNop().Sugar(), tally.NoopScope, registry, errs.NewClassifierProcessor(), gatenoop.New()) + require.NoError(t, runtime.Register(observer)) + t.Cleanup(func() { require.NoError(t, runtime.Stop(30000)) }) + + for _, notification := range []struct { + id string + final bool + }{{"final", true}, {"late-refresh", false}} { + payload, err := mq.Marshal(&mq.IssueTrack{QueueName: queueName, BuildId: buildID, Finalize: notification.final}) + require.NoError(t, err) + require.NoError(t, publish.Message(ctx, registry, mq.TopicKeyIssueTrack, publish.MessageParams{Tenant: queueName, ID: notification.id, PartitionKey: queueName, Payload: payload})) + } + require.NoError(t, runtime.Start(ctx)) + + completed := map[string]bool{} + held := false + for len(completed) != 2 { + result := <-results + require.NoError(t, result.err) + held = held || result.held + if !result.held { + require.Equal(t, entity.IssueTrackingPhaseFinalized, result.progress.Phase) + completed[result.id] = true + } + } + require.True(t, held) + issues, err := bound.GetIssueStateStore().List(ctx, storage.IssueStateRange{Limit: 200}) + require.NoError(t, err) + require.Len(t, issues, 104) + open := 0 + for _, issue := range issues { + if issue.Status == entity.IssueStatusOpen { + open++ + } + } + require.Equal(t, 103, open) + closed, err := bound.GetIssueStateStore().Get(ctx, "z", "") + require.NoError(t, err) + require.Equal(t, entity.IssueStatusClosed, closed.Status) + require.Equal(t, int32(2), closed.Version) + _, err = bound.GetIssueLogStore().Get(ctx, "z", "", buildID, entity.IssueEventResolved) + require.NoError(t, err) + checkpoint, err := bound.GetIssueTrackingStore().Get(ctx) + require.NoError(t, err) + require.Equal(t, "final", checkpoint.NotificationID) +} + +type pausedCreateFactory struct { + storage.Factory + paused chan struct{} + resume <-chan struct{} +} + +func (f pausedCreateFactory) For(config storage.Config) (storage.Storage, error) { + bound, err := f.Factory.For(config) + if err != nil { + return nil, err + } + return pausedCreateStorage{bound, pausedCreateIssueStore{bound.GetIssueStateStore(), f.paused, f.resume}}, nil +} + +type pausedCreateStorage struct { + storage.Storage + issues storage.IssueStateStore +} + +func (s pausedCreateStorage) GetIssueStateStore() storage.IssueStateStore { return s.issues } + +type pausedCreateIssueStore struct { + storage.IssueStateStore + paused chan struct{} + resume <-chan struct{} +} + +func (s pausedCreateIssueStore) Create(ctx context.Context, issue entity.IssueState) error { + s.paused <- struct{}{} + <-s.resume + return s.IssueStateStore.Create(ctx, issue) +} + +func testIssueTrackingLeaseTakeover(t *testing.T, ctx context.Context, db *sql.DB, factory storage.Factory) { + const queueName = "tracking/takeover" + const topic = "tracking_takeover" + const group = "tracking-takeover" + bound, err := factory.For(storage.Config{QueueName: queueName}) + require.NoError(t, err) + for _, source := range []struct { + id string + status entity.BuildStatus + }{{"100", entity.BuildStatusRunning}, {"101", entity.BuildStatusSucceeded}} { + request := entity.Request{ID: source.id, Queue: queueName, URI: "git://tracking/" + source.id, State: entity.RequestStateProcessing, Version: 1} + require.NoError(t, bound.GetRequestStore().Create(ctx, request)) + if source.status.IsTerminal() { + request.State, request.TerminalBuildID = entity.RequestStateSucceeded, "build/"+source.id + require.NoError(t, bound.GetRequestStore().Update(ctx, request, 1, 2)) + } + require.NoError(t, bound.GetBuildStore().Create(ctx, entity.Build{ID: "build/" + source.id, RequestID: source.id, Status: source.status, Version: 1})) + } + require.NoError(t, bound.GetBuildIssueStore().Create(ctx, "build/100", entity.BuildIssue{TaskID: "task", UnitID: "unit"})) + testutil.ApplySchema(t, testutil.NewTestLogger(t), db, testutil.SchemaDir("platform/extension/messagequeue/mysql/schema")) + paused := make(chan struct{}, 1) + resume := make(chan struct{}) + aFactory := pausedCreateFactory{factory, paused, resume} + newRuntime := func(name string, stores storage.Factory) (consumer.Consumer, consumer.TopicRegistry, <-chan trackingDeliveryResult) { + q, err := queuemysql.NewQueue(queuemysql.Params{DB: db, Logger: zap.NewNop(), MetricsScope: tally.NoopScope, Tenants: []string{queueName}}) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, q.Close()) }) + config := extqueue.DefaultSubscriptionConfig(name, group) + config.PollIntervalMs = 20 + config.PartitionDiscoveryIntervalMs = 50 + config.LeaseRenewalIntervalMs = 50 + config.LeaseDurationMs = 1000 + config.VisibilityTimeoutMs = 200 + config.BatchSize = 1 + config.Retry.MaxAttempts = 0 + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{{Key: mq.TopicKeyIssueTrack, Name: topic, Queue: q, Subscription: config}}) + require.NoError(t, err) + results := make(chan trackingDeliveryResult, 64) + controller := issuetrack.NewController(zap.NewNop().Sugar(), tally.NoopScope, stores, mq.TopicKeyIssueTrack, group) + observer := &trackingObserver{Controller: controller, progress: bound.GetIssueTrackingStore(), results: results} + runtime := consumer.New(zap.NewNop().Sugar(), tally.NoopScope, registry, errs.NewClassifierProcessor(), gatenoop.New()) + require.NoError(t, runtime.Register(observer)) + t.Cleanup(func() { require.NoError(t, runtime.Stop(30000)) }) + return runtime, registry, results + } + a, registry, resultsA := newRuntime("tracking-a", aFactory) + b, _, resultsB := newRuntime("tracking-b", factory) + publishNotice := func(buildID, id string, final bool) { + payload, err := mq.Marshal(&mq.IssueTrack{QueueName: queueName, BuildId: buildID, Finalize: final}) + require.NoError(t, err) + require.NoError(t, publish.Message(ctx, registry, mq.TopicKeyIssueTrack, publish.MessageParams{Tenant: queueName, ID: id, PartitionKey: queueName, Payload: payload})) + } + publishNotice("build/100", "old-refresh", false) + require.NoError(t, a.Start(ctx)) + <-paused + defer close(resume) + require.NoError(t, b.Start(ctx)) + // Transfer the partition while A is blocked at its real SQL create. + result, err := db.ExecContext(ctx, "UPDATE queue_partition_leases SET leased_by = ?, lease_renewed_at = ? WHERE tenant = ? AND consumer_group = ? AND topic = ? AND partition_key = ?", "tracking-b", time.Now().UnixMilli(), queueName, group, topic, queueName) + require.NoError(t, err) + rows, err := result.RowsAffected() + require.NoError(t, err) + require.EqualValues(t, 1, rows) + resultB := <-resultsB + require.NoError(t, resultB.err) + require.Equal(t, "build/100", resultB.progress.BuildID) + publishNotice("build/101", "new-final", true) + for { + resultB = <-resultsB + require.NoError(t, resultB.err) + if resultB.id == "new-final" { + require.Equal(t, entity.IssueTrackingPhaseFinalized, resultB.progress.Phase) + break + } + } + closed, err := bound.GetIssueStateStore().Get(ctx, "task", "unit") + require.NoError(t, err) + require.Equal(t, entity.IssueStatusClosed, closed.Status) + resume <- struct{}{} + resultA := <-resultsA + require.Error(t, resultA.err) + issue, err := bound.GetIssueStateStore().Get(ctx, "task", "unit") + require.NoError(t, err) + require.Equal(t, closed, issue) + require.Equal(t, entity.IssueStatusClosed, issue.Status) + require.Equal(t, int32(2), issue.Version) + _, err = bound.GetIssueLogStore().Get(ctx, "task", "unit", "build/100", entity.IssueEventOpened) + require.NoError(t, err) + _, err = bound.GetIssueLogStore().Get(ctx, "task", "unit", "build/101", entity.IssueEventResolved) + require.NoError(t, err) +} diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index 514bf5b06..abf5e55a1 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -115,6 +115,8 @@ func TestMySQLStorage(t *testing.T) { testIssueStateList(t, ctx, factory) }) + t.Run("IssueTrackingDelivery", func(t *testing.T) { resetStorage(t, db); testIssueTrackingDelivery(t, ctx, db, factory) }) + t.Run("IssueTrackingLeaseTakeover", func(t *testing.T) { resetStorage(t, db); testIssueTrackingLeaseTakeover(t, ctx, db, factory) }) t.Run("IssueTrackingStore", func(t *testing.T) { resetStorage(t, db) testIssueTrackingStore(t, ctx, factory)