diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 6e3f98b1d..509122734 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -567,6 +567,11 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe Name: "issuetrack", Queue: q, }, + { + Key: stovepipemq.TopicKeyFix, + Name: "fix", + Queue: q, + }, { Key: stovepipemq.TopicKeyBuildAnalysis, Name: "buildanalysis", diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go index 3f458b896..6dd54825e 100644 --- a/service/stovepipe/server/main_test.go +++ b/service/stovepipe/server/main_test.go @@ -101,6 +101,14 @@ func TestIssueTrackingHasPublicationRouteWithoutConsumption(t *testing.T) { assert.NotContains(t, topicKeys(deadLetter), dlq.TopicKey(stovepipemq.TopicKeyIssueTrack)) } +func TestFixHasPublicationRouteWithoutConsumption(t *testing.T) { + registry, primary, _ := registeredControllers(t) + name, ok := registry.TopicName(stovepipemq.TopicKeyFix) + require.True(t, ok) + assert.Equal(t, "fix", name) + assert.NotContains(t, topicKeys(primary), stovepipemq.TopicKeyFix) +} + func TestHookStage(t *testing.T) { registry, primary, deadLetter := registeredControllers(t) diff --git a/stovepipe/controller/fix/README.md b/stovepipe/controller/fix/README.md index bca017ade..489766326 100644 --- a/stovepipe/controller/fix/README.md +++ b/stovepipe/controller/fix/README.md @@ -4,6 +4,6 @@ The controller consumes a `FixRequest` containing a queue name and build ID. It A returned fix ID is associated with the issue through `BuildIssueStore.Update`, using the version read during selection. A concurrent association is preserved. A submission error without an ID leaves the issue eligible for normal queue retries; an ID returned alongside an error is retained. -Issue tracking calls [`fixrequest.Publish`](../../core/fixrequest/publish.go) after persisting both the build observation and open issue. Publications for the same build share a message identity and partition. +Issue tracking calls [`fixrequest.Publish`](../../core/fixrequest/publish.go) after completing a nonempty opening pass and persisting its issue state and history. Publications share a build partition, while the message identity includes the pass notification ID so a later pass can trigger another evaluation. A pass with no eligible issue completes without submitting a repair; a recorded fix ID prevents later passes from submitting another repair for the build. The one-per-build limit is controller policy. Submission precedes persistence, so overlapping deliveries, crashes, and failed writes can still produce duplicate executor calls. Issue recovery remains separate from repair association. diff --git a/stovepipe/controller/issuetrack/BUILD.bazel b/stovepipe/controller/issuetrack/BUILD.bazel index 663d00d00..f8d6359fe 100644 --- a/stovepipe/controller/issuetrack/BUILD.bazel +++ b/stovepipe/controller/issuetrack/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//stovepipe/core/fixrequest:go_default_library", "//stovepipe/core/loader:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", @@ -23,13 +24,16 @@ go_test( name = "go_default_test", srcs = [ "delivery_test.go", + "fix_handoff_test.go", "fixture_test.go", "issuetrack_test.go", ], embed = [":go_default_library"], deps = [ "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", "//platform/consumer/mock:go_default_library", + "//platform/extension/messagequeue/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 e3668a164..abd4fb740 100644 --- a/stovepipe/controller/issuetrack/README.md +++ b/stovepipe/controller/issuetrack/README.md @@ -22,6 +22,8 @@ The post-read cursor write and processed-build updates, including unchanged stat Every pass starts in `refreshing`, applying the build's accumulated issues in primary-key order. Reported issues are opened or kept open; nothing is closed. A terminal notification retains `Finalize = true` in the progress record, so a replacement worker can distinguish terminal work from an ordinary refresh even when helping it through another notification. Encountering a task-wide issue (empty unit ID) sets `HasTaskWideIssues` in that same versioned record. This fact remains set across batches and retries, including when the issue was already applied by an earlier pass. +Once the entire build-issue range is exhausted, an opening pass with a nonempty task cursor publishes a build-scoped repair request before saving its next phase. Empty passes and closing sweeps do not publish. Publication failures leave the pass active so retries repeat the handoff after ensuring issue state and history are durable. The repair message identity combines the build and persisted pass notification IDs: retries share an identity, while later passes can trigger another evaluation. The fix controller selects an eligible issue and submits at most one repair per build. + Once the entire build-issue range is exhausted: - An ordinary refresh becomes `refreshed`. @@ -42,7 +44,7 @@ Build IDs are opaque; source order comes from Request IDs. All scans are bounded ## 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. +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. The host must provide a fix publication route and a repair executor when registering the fix consumer. ## Analysis handoff diff --git a/stovepipe/controller/issuetrack/fix_handoff_test.go b/stovepipe/controller/issuetrack/fix_handoff_test.go new file mode 100644 index 000000000..1b82ea3bc --- /dev/null +++ b/stovepipe/controller/issuetrack/fix_handoff_test.go @@ -0,0 +1,197 @@ +// 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 ( + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/entity" +) + +func TestOpeningPassPublishesOneBuildRepairRequest(t *testing.T) { + for _, tt := range []struct { + name string + status entity.BuildStatus + final bool + }{ + {name: "running refresh", status: entity.BuildStatusRunning}, + {name: "terminal opening pass", status: entity.BuildStatusFailed, final: true}, + {name: "canceled build", status: entity.BuildStatusCancelled, final: true}, + } { + t.Run(tt.name, func(t *testing.T) { + f := newFixture(t) + f.source("build", "10", tt.status) + f.seedIssue("reopened", "unit", entity.IssueStatusClosed) + for _, task := range []string{"new", "reopened"} { + f.reportIssue("build", task, "unit") + } + f.onFixPublish = func(msg entityqueue.Message) { + require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase) + for _, task := range []string{"new", "reopened"} { + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{task, "unit"}].Status) + require.Contains(t, f.logs, logKey{issueKey{task, "unit"}, "build", entity.IssueEventOpened}) + } + payload := &stovepipemq.FixRequest{} + require.NoError(t, stovepipemq.Unmarshal(msg.Payload, payload)) + require.Equal(t, "build", payload.GetId()) + require.Equal(t, testQueue, payload.GetQueueName()) + require.Equal(t, testQueue, msg.Tenant) + require.Equal(t, "build", msg.PartitionKey) + } + msg := f.message("build", tt.final) + closing := tt.final && tt.status != entity.BuildStatusCancelled + require.NoError(t, f.deliver(msg, closing)) + require.Len(t, f.fixMessages, 1) + if closing { + require.NoError(t, f.deliver(msg, false)) + require.Len(t, f.fixMessages, 1) + } + require.NoError(t, f.deliver(msg, false)) + require.Len(t, f.fixMessages, 1) + }) + } +} + +func TestRepairHandoffSkipsEmptyPassAndPublishesLaterPasses(t *testing.T) { + for _, tt := range []struct { + name string + reportExisting bool + initialPublishes int + }{ + {name: "empty pass"}, + {name: "previously open issues", reportExisting: true, initialPublishes: 1}, + } { + t.Run(tt.name, func(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusRunning) + f.seedIssue("existing", "unit", entity.IssueStatusOpen) + if tt.reportExisting { + f.reportIssue("build", "existing", "unit") + } + require.NoError(t, f.deliver(f.message("build", false), false)) + require.Len(t, f.fixMessages, tt.initialPublishes) + f.reportIssue("build", "new", "unit") + require.NoError(t, f.deliver(f.message("build", false), false)) + require.Len(t, f.fixMessages, tt.initialPublishes+1) + if tt.initialPublishes != 0 { + require.NotEqual(t, f.fixMessages[0].ID, f.fixMessages[1].ID) + require.Equal(t, f.fixMessages[0].PartitionKey, f.fixMessages[1].PartitionKey) + } + }) + } +} + +func TestRepairHandoffWaitsForAllOpeningBatches(t *testing.T) { + for _, issueCount := range []int{_issueBatchSize, _issueBatchSize + 1} { + t.Run(fmt.Sprintf("%d issues", issueCount), func(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusRunning) + f.reportIssue("build", "a-new", "unit") + for i := 1; i < issueCount; i++ { + unit := fmt.Sprintf("unit/%03d", i) + f.seedIssue("existing", unit, entity.IssueStatusOpen) + f.reportIssue("build", "existing", unit) + } + msg := f.message("build", false) + require.NoError(t, f.deliver(msg, true)) + require.Empty(t, f.fixMessages) + + f.onFixPublish = func(entityqueue.Message) { + for _, issue := range f.states { + require.Equal(t, "build", issue.LastProcessedBuildID) + } + } + require.NoError(t, f.deliver(msg, false)) + require.Len(t, f.fixMessages, 1) + }) + } +} + +func TestRepairHandoffRetriesBeforeCompletingPass(t *testing.T) { + for _, tt := range []struct { + name string + failure string + failOnPhase entity.IssueTrackingPhase + initialPublishes int + }{ + {name: "issue write", failure: "issue/create"}, + {name: "opening history", failure: "history/create"}, + {name: "repair publication", failure: "fix/publish", initialPublishes: 1}, + {name: "completion checkpoint", failOnPhase: entity.IssueTrackingPhaseRefreshed, initialPublishes: 1}, + } { + t.Run(tt.name, func(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusRunning) + f.reportIssue("build", "new", "unit") + f.failures[tt.failure] = 1 + f.failOnPhase = tt.failOnPhase + msg := f.message("build", false) + require.ErrorIs(t, f.deliver(msg, false), assert.AnError) + require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase) + require.Len(t, f.fixMessages, tt.initialPublishes) + + require.NoError(t, f.deliver(msg, false)) + require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) + require.Len(t, f.fixMessages, tt.initialPublishes+1) + require.Len(t, f.logs, 1) + if tt.initialPublishes != 0 { + require.Equal(t, f.fixMessages[0].ID, f.fixMessages[1].ID) + require.Equal(t, f.fixMessages[0].Payload, f.fixMessages[1].Payload) + } + }) + } +} + +func TestRepairHandoffUsesResumedBuildAndPassIdentity(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusRunning) + f.source("next-build", "11", entity.BuildStatusRunning) + f.reportIssue("build", "new", "unit") + f.failures["fix/publish"] = 1 + msg := f.message("build", false) + require.ErrorIs(t, f.deliver(msg, false), assert.AnError) + + nextMessage := f.message("next-build", false) + require.NoError(t, f.deliver(nextMessage, true)) + require.Equal(t, "build", f.progress.BuildID) + require.Len(t, f.fixMessages, 2) + require.Equal(t, f.fixMessages[0].ID, f.fixMessages[1].ID) + require.Equal(t, "build", f.fixMessages[1].PartitionKey) + + require.NoError(t, f.deliver(msg, false)) + require.Len(t, f.fixMessages, 2) + + f.reportIssue("next-build", "new", "unit") + require.NoError(t, f.deliver(nextMessage, false)) + require.Len(t, f.fixMessages, 3) + require.NotEqual(t, f.fixMessages[0].ID, f.fixMessages[2].ID) + require.Equal(t, "next-build", f.fixMessages[2].PartitionKey) +} + +func TestTaskWideFailureStillPublishesRepair(t *testing.T) { + f := newFixture(t) + f.source("build", "10", entity.BuildStatusFailed) + f.seedIssue("existing", "unit", entity.IssueStatusOpen) + f.reportIssue("build", "setup", "") + require.NoError(t, f.deliver(f.message("build", true), false)) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"existing", "unit"}].Status) + require.Len(t, f.fixMessages, 1) +} diff --git a/stovepipe/controller/issuetrack/fixture_test.go b/stovepipe/controller/issuetrack/fixture_test.go index 6db9a8c3c..929067ed0 100644 --- a/stovepipe/controller/issuetrack/fixture_test.go +++ b/stovepipe/controller/issuetrack/fixture_test.go @@ -22,6 +22,9 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" @@ -55,6 +58,9 @@ type fixture struct { requests map[string]entity.Request buildIssues map[string]map[issueKey]entity.BuildIssue sequence int + + fixMessages []entityqueue.Message + onFixPublish func(entityqueue.Message) } func newFixture(t *testing.T) *fixture { @@ -235,7 +241,19 @@ func newFixture(t *testing.T) *fixture { } return result, nil }).AnyTimes() - f.controller = NewController(zap.NewNop().Sugar(), tally.NoopScope, f.factory, messagequeue.TopicKeyIssueTrack, "test") + queue := queuemock.NewMockQueue(f.mocks) + publisher := queuemock.NewMockPublisher(f.mocks) + queue.EXPECT().Publisher().Return(publisher).AnyTimes() + publisher.EXPECT().Publish(gomock.Any(), "fix", gomock.Any()).DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { + f.fixMessages = append(f.fixMessages, msg) + if f.onFixPublish != nil { + f.onFixPublish(msg) + } + return f.fail("fix/publish") + }).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{{Key: messagequeue.TopicKeyFix, Name: "fix", Queue: queue}}) + require.NoError(t, err) + f.controller = NewController(zap.NewNop().Sugar(), tally.NoopScope, f.factory, registry, messagequeue.TopicKeyIssueTrack, "test") f.controller.nowMs = func() int64 { return 1000 } return f } diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go index 8902fdc2f..a39576d01 100644 --- a/stovepipe/controller/issuetrack/issuetrack.go +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -27,6 +27,7 @@ import ( 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/fixrequest" "github.com/uber/submitqueue/stovepipe/core/loader" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" @@ -44,16 +45,18 @@ type Controller struct { logger *zap.SugaredLogger metricsScope tally.Scope stores storage.Factory + registry consumer.TopicRegistry 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 { +func NewController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string) *Controller { return &Controller{ logger: logger.Named("issuetrack_controller"), metricsScope: scope.SubScope("issuetrack_controller"), stores: stores, + registry: registry, topicKey: topicKey, consumerGroup: consumerGroup, nowMs: func() int64 { return time.Now().UnixMilli() }, @@ -246,6 +249,11 @@ func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage, } if len(keys) < limit { + if !finalizing && progress.CursorTaskID != "" { + if err := fixrequest.Publish(ctx, c.registry, queue, progress.BuildID, progress.NotificationID); err != nil { + return entity.IssueTrackingState{}, fmt.Errorf("publish build repair request: %w", err) + } + } next := progress switch { case finalizing: diff --git a/stovepipe/core/fixrequest/publish.go b/stovepipe/core/fixrequest/publish.go index 21f684316..f41a6147f 100644 --- a/stovepipe/core/fixrequest/publish.go +++ b/stovepipe/core/fixrequest/publish.go @@ -24,11 +24,11 @@ import ( mq "github.com/uber/submitqueue/stovepipe/core/messagequeue" ) -// Publish requires the build and its open issues to be persisted first. -// Replays after publication errors must call Publish again. -func Publish(ctx context.Context, registry consumer.TopicRegistry, queueName, buildID string) error { - if queueName == "" || buildID == "" { - return fmt.Errorf("repair request requires a queue and build ID") +// Publish requires the opening pass's issue state and history to be durable. +// The notification ID must be stable across retries and distinct between passes. +func Publish(ctx context.Context, registry consumer.TopicRegistry, queueName, buildID, notificationID string) error { + if queueName == "" || buildID == "" || notificationID == "" { + return fmt.Errorf("repair request requires a queue, build ID, and notification ID") } payload, err := mq.Marshal(&mq.FixRequest{Id: buildID, QueueName: queueName}) @@ -36,10 +36,9 @@ func Publish(ctx context.Context, registry consumer.TopicRegistry, queueName, bu return fmt.Errorf("failed to serialize repair request: %w", err) } - // Repair submission is a one-shot handoff for this build. return publish.Message(ctx, registry, mq.TopicKeyFix, publish.MessageParams{ Tenant: queueName, - ID: publish.IntentID(buildID), + ID: publish.IntentID(buildID, notificationID), Payload: payload, PartitionKey: buildID, }) diff --git a/stovepipe/core/fixrequest/publish_test.go b/stovepipe/core/fixrequest/publish_test.go index 0b2a600cc..ee5146246 100644 --- a/stovepipe/core/fixrequest/publish_test.go +++ b/stovepipe/core/fixrequest/publish_test.go @@ -40,7 +40,7 @@ func newTestRegistry(t *testing.T) (consumer.TopicRegistry, *queuemock.MockPubli return registry, publisher } -func TestPublishBuildIdentityAndRetry(t *testing.T) { +func TestPublishBuildAndPassIdentityAndRetry(t *testing.T) { registry, publisher := newTestRegistry(t) var messages []entityqueue.Message publisher.EXPECT().Publish(gomock.Any(), "test-fix", gomock.Any()). @@ -50,18 +50,22 @@ func TestPublishBuildIdentityAndRetry(t *testing.T) { return assert.AnError } return nil - }).Times(4) + }).Times(5) ctx := context.Background() - require.ErrorIs(t, Publish(ctx, registry, "validation/main", "Build/42 "), assert.AnError) - require.NoError(t, Publish(ctx, registry, "validation/main", "Build/42 ")) - require.NoError(t, Publish(ctx, registry, "validation/main", "Build/43")) - require.NoError(t, Publish(ctx, registry, "validation/other", "Build/42 ")) + require.ErrorIs(t, Publish(ctx, registry, "validation/main", "Build/42 ", "pass/1"), assert.AnError) + require.NoError(t, Publish(ctx, registry, "validation/main", "Build/42 ", "pass/1")) + require.NoError(t, Publish(ctx, registry, "validation/main", "Build/42 ", "pass/2")) + require.NoError(t, Publish(ctx, registry, "validation/main", "Build/43", "pass/1")) + require.NoError(t, Publish(ctx, registry, "validation/other", "Build/42 ", "pass/1")) assert.Equal(t, messages[0].ID, messages[1].ID) assert.Equal(t, messages[0].PartitionKey, messages[1].PartitionKey) assert.NotEqual(t, messages[1].ID, messages[2].ID) - assert.NotEqual(t, messages[1].Tenant, messages[3].Tenant) + assert.Equal(t, messages[1].PartitionKey, messages[2].PartitionKey) + assert.Equal(t, messages[1].Payload, messages[2].Payload) + assert.NotEqual(t, messages[1].ID, messages[3].ID) + assert.NotEqual(t, messages[1].Tenant, messages[4].Tenant) for _, message := range messages { payload := &mq.FixRequest{} require.NoError(t, mq.Unmarshal(message.Payload, payload)) @@ -74,22 +78,24 @@ func TestPublishBuildIdentityAndRetry(t *testing.T) { func TestPublishRejectsMissingIdentity(t *testing.T) { for _, test := range []struct { - name string - queue string - build string + name string + queue string + build string + notification string }{ - {name: "missing queue", build: "build/42"}, - {name: "missing build", queue: "validation/main"}, + {name: "missing queue", build: "build/42", notification: "pass/1"}, + {name: "missing build", queue: "validation/main", notification: "pass/1"}, + {name: "missing notification", queue: "validation/main", build: "build/42"}, } { t.Run(test.name, func(t *testing.T) { registry, _ := newTestRegistry(t) - require.Error(t, Publish(context.Background(), registry, test.queue, test.build)) + require.Error(t, Publish(context.Background(), registry, test.queue, test.build, test.notification)) }) } } func TestPublishReturnsConfigurationAndSerializationErrors(t *testing.T) { - require.Error(t, Publish(context.Background(), consumer.TopicRegistry{}, "validation/main", "build/42")) + require.Error(t, Publish(context.Background(), consumer.TopicRegistry{}, "validation/main", "build/42", "pass/1")) registry, _ := newTestRegistry(t) - require.Error(t, Publish(context.Background(), registry, "validation/main", string([]byte{0xff}))) + require.Error(t, Publish(context.Background(), registry, "validation/main", string([]byte{0xff}), "pass/1")) }