diff --git a/stovepipe/controller/issuetrack/BUILD.bazel b/stovepipe/controller/issuetrack/BUILD.bazel index 441031a49..663d00d00 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", "fixture_test.go", "issuetrack_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 3af6e37f9..ab316bc3a 100644 --- a/stovepipe/controller/issuetrack/README.md +++ b/stovepipe/controller/issuetrack/README.md @@ -1,6 +1,6 @@ # 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. +Issue tracking maintains current issue state and opening/closing history from persisted build issues. Each delivery handles a bounded batch of issues and requests immediate redelivery when more work remains. No consumer is registered yet. ## Applying one issue @@ -17,3 +17,17 @@ Closed issue rows are retained, so their versions cannot reset on reopening. A r 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. + +## Bounded passes + +Every delivery processes at most 100 task/unit keys, including a revisited cursor and retained closed rows. Refreshes open reported issues without closing missing ones. Finalization also scans issue states and closes issues absent from that build's accumulated issues. A canceled build may not have performed validation, so it can open reported issues but cannot close existing issues by absence. + +The cursor is inclusive: revisit it to finish the issue and its history, then continue the ordered range after it. A full batch retains the active phase and requests redelivery. Exhausting the range conditionally marks the pass refreshed or finalized. + +An unfinished pass completes before another source is admitted. A notification that helps finish a different pass requests redelivery before handling its own build. Completed notifications and older sources are ignored, as are refreshes after the same Request has finalized. A final notification requires a terminal build chosen by its Request. + +Build IDs are opaque; source order comes from Request IDs. All scans are bounded reads under leading primary-key prefixes. There is no transaction spanning progress, issue state, and history. + +## Recovery and activation + +An error returns to normal consumer retry. The next delivery resumes the cursor before advancing. Permanently failing cursor work blocks issue tracking for that queue; terminal recovery and consumer registration remain follow-up work. This PR does not enable consumption or publish repair requests. diff --git a/stovepipe/controller/issuetrack/delivery_test.go b/stovepipe/controller/issuetrack/delivery_test.go new file mode 100644 index 000000000..36c6e130e --- /dev/null +++ b/stovepipe/controller/issuetrack/delivery_test.go @@ -0,0 +1,409 @@ +// 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" + "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" +) + +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(8), f.states[issueKey{"b", "unit"}].Version) + require.Equal(t, "build", f.states[issueKey{"c", "unit"}].FirstSeenBuildID) + require.Equal(t, int32(8), 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.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 TestAnotherNotificationFinishesTheCursorBeforeItsOwnBuild(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(currentCursor()) + 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 TestIssueMutationWaitsForItsCursor(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.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) + require.Equal(t, "task", f.progress.CursorTaskID) + require.Equal(t, "unit", f.progress.CursorUnitID) + require.NoError(t, f.deliver(msg, false)) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + 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, int64(1000), 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.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.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 TestReadFailuresDoNotChangeIssues(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) + }) + } +} + +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 TestCursorRetriesPartialWrites(t *testing.T) { + for _, operation := range []string{"issue/create", "issue/update", "history/create", "complete-pass"} { + t.Run(operation, func(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusRunning) + f.reportIssue("build", "task", "unit") + if operation == "issue/update" { + f.seedIssue("task", "unit", entity.IssueStatusClosed) + } + if operation == "complete-pass" { + f.failOnPhase = entity.IssueTrackingPhaseRefreshed + } else { + f.failures[operation] = 1 + } + msg := f.message("build", false) + require.ErrorIs(t, f.deliver(msg, false), assert.AnError) + require.Equal(t, "task", f.progress.CursorTaskID) + require.Equal(t, "unit", f.progress.CursorUnitID) + require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase) + require.NoError(t, f.deliver(msg, false)) + require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) + require.Equal(t, "build", f.states[issueKey{"task", "unit"}].LastProcessedBuildID) + require.Equal(t, "build", f.states[issueKey{"task", "unit"}].LastChangedBuildID) + require.Len(t, f.logs, 1) + if operation == "issue/update" { + require.Equal(t, int32(8), f.states[issueKey{"task", "unit"}].Version) + } else { + require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version) + } + }) + } +} + +func TestHistoryRecordsTheActualChangingBuildNotTheLatestProcessedBuild(t *testing.T) { + f := newFixture(t) + f.seedIssue("task", "unit", entity.IssueStatusOpen) + f.source("build/100", "100", entity.BuildStatusSucceeded) + f.failures["history/create"] = 1 + msg := f.message("build/100", true) + require.ErrorIs(t, f.deliver(msg, false), assert.AnError) + changed := f.states[issueKey{"task", "unit"}] + require.Equal(t, entity.IssueStatusClosed, changed.Status) + require.Equal(t, "build/100", changed.LastChangedBuildID) + require.Equal(t, "build/100", changed.LastProcessedBuildID) + require.Equal(t, "older", changed.FirstSeenBuildID) + require.Empty(t, f.logs) + f.controller.nowMs = func() int64 { return 2000 } + require.NoError(t, f.deliver(msg, false)) + require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventResolved}].TimestampMs) + require.Equal(t, changed, f.states[issueKey{"task", "unit"}]) + + f.source("build/101", "101", entity.BuildStatusSucceeded) + require.NoError(t, f.deliver(f.message("build/101", true), false)) + retained := f.states[issueKey{"task", "unit"}] + require.Equal(t, "build/101", retained.LastProcessedBuildID) + require.Equal(t, "build/100", retained.LastChangedBuildID) + require.Equal(t, int64(1000), retained.ChangedAtMs) + require.Equal(t, int32(9), retained.Version) + require.Len(t, f.logs, 1) +} diff --git a/stovepipe/controller/issuetrack/fixture_test.go b/stovepipe/controller/issuetrack/fixture_test.go index cd24c8f63..6db9a8c3c 100644 --- a/stovepipe/controller/issuetrack/fixture_test.go +++ b/stovepipe/controller/issuetrack/fixture_test.go @@ -39,6 +39,7 @@ type logKey struct { } type fixture struct { + failOnPhase entity.IssueTrackingPhase t *testing.T mocks *gomock.Controller controller *Controller @@ -146,6 +147,10 @@ func newFixture(t *testing.T) *fixture { if err := f.fail("progress/update"); err != nil { return err } + if f.failOnPhase != "" && state.Phase == f.failOnPhase { + f.failOnPhase = "" + return assert.AnError + } if !f.hasProgress || f.progress.Version != oldVersion { return storage.ErrVersionMismatch } diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go index 72cf7b0b9..3096b21a4 100644 --- a/stovepipe/controller/issuetrack/issuetrack.go +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -20,16 +20,24 @@ 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 processes at most this many keys, including a revisited cursor and closed records. +const _issueBatchSize = 100 + type issueKey struct{ task, unit string } // Controller retains issue transitions and their queue-scoped progress. @@ -53,6 +61,213 @@ func NewController(logger *zap.SugaredLogger, scope tally.Scope, stores storage. } } +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 + + 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") + } + + if progress.CursorTaskID != "" { + next, err := c.processIssue(ctx, store, progress, issueKey{progress.CursorTaskID, progress.CursorUnitID}, false) + if err != nil { + return entity.IssueTrackingState{}, err + } + progress = next + limit-- + } + + 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] + } + + cursorNeedsCheckpoint := false + for _, key := range keys { + if err := ctx.Err(); err != nil { + return entity.IssueTrackingState{}, err + } + previousVersion := progress.Version + progress, err = c.processIssue(ctx, store, progress, key, reported[key]) + if err != nil { + return entity.IssueTrackingState{}, err + } + cursorNeedsCheckpoint = progress.Version == previousVersion + } + + if complete { + next := progress + next.Phase = entity.IssueTrackingPhaseRefreshed + if finalizing { + next.Phase = entity.IssueTrackingPhaseFinalized + } + return saveProgress(ctx, store.GetIssueTrackingStore(), next) + } + if cursorNeedsCheckpoint { + return saveProgress(ctx, store.GetIssueTrackingStore(), progress) + } + + return progress, nil +} + 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 @@ -182,6 +397,26 @@ func saveProgress(ctx context.Context, store storage.IssueTrackingStore, next en 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)