From 00d3db97666c6692912f9cfc04adde670cf80844 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Thu, 29 Oct 2026 14:37:55 +0000 Subject: [PATCH] feat(stovepipe): enable fix consumer with no-op repairs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Register the fix consumer in the reference host with an extended retry subscription. Use a no-op repair executor that returns `noop/` without launching work. The fix controller records the usual one-per-build association. Fix dead-letter recovery consumer registration remains separate. ## Test Plan Run ./tool/bazel test //service/stovepipe/server:go_default_test //stovepipe/controller/issuetrack:go_default_test //stovepipe/controller/fix:go_default_test //stovepipe/core/fixrequest:go_default_test //test/integration/stovepipe:go_default_test --test_output=errors. The Docker/MySQL integration suite starts the reference server with the new subscription, and the wiring test exercises no-op acceptance through the registered fix controller. ## Revert Plan Revert this PR to stop the fix consumer and remove the no-op executor. No schema changes are required. ## Issue Links None — reference-host fix wiring with no associated ticket. --- service/stovepipe/README.md | 4 +- service/stovepipe/server/BUILD.bazel | 9 +++ service/stovepipe/server/main.go | 26 +++++-- service/stovepipe/server/main_test.go | 81 ++++++++++++++++++++-- stovepipe/controller/issuetrack/README.md | 2 +- stovepipe/extension/fixer/noop/BUILD.bazel | 12 ++++ stovepipe/extension/fixer/noop/noop.go | 34 +++++++++ 7 files changed, 157 insertions(+), 11 deletions(-) create mode 100644 stovepipe/extension/fixer/noop/BUILD.bazel create mode 100644 stovepipe/extension/fixer/noop/noop.go diff --git a/service/stovepipe/README.md b/service/stovepipe/README.md index b447b64fa..639aec723 100644 --- a/service/stovepipe/README.md +++ b/service/stovepipe/README.md @@ -11,6 +11,7 @@ Runnable wiring for the **Stovepipe** domain — a single-service domain (the do - **buildsignal consumer** (`TopicKeyBuildSignal`) — polls/records the build's terminal status and releases the queue's in-flight slot, then publishes to `record`. - **buildanalysis consumer** (`TopicKeyBuildAnalysis`) — periodically detects and persists build issues, then publishes queue-partitioned issue-tracking notifications. - **issuetrack consumer** (`TopicKeyIssueTrack`) — applies reported build issues to current issue state and history, closing absent issues only when final reconciliation permits it. +- **fix consumer** (`TopicKeyFix`) — selects an eligible issue and records one repair association per build using the reference host's no-op executor. - **record consumer** (`TopicKeyRecord`) — writes the whole-repository validation fact and, for a green fact, advances the queue's last-green bookmark and promotion ref. - **hook consumer** (`TopicKeyHook`) — receives `validation.repository.started` from process and `validation.repository.recorded` or `validation.repository.cancelled` from record. The current resolver invokes only `noop`, so these events have no external side effect. Its topic name is domain-qualified (`stovepipe-hook`) because the key is shared across domains. - **DLQ consumers** — registered for process, build, buildsignal, buildanalysis, issuetrack, record, and hook, with stage-specific recovery behavior. Issue-tracking recovery abandons incomplete passes while preserving accepted changes; it does not rerun their remaining batches. @@ -21,10 +22,11 @@ Stovepipe therefore needs two MySQL databases: a **storage** database (the `queu ## Wiring notes -`server/main.go` is the composition root and supplies the concrete extension implementations. Two are deliberately demo-only and must be replaced for any real deployment: +`server/main.go` is the composition root and supplies the concrete extension implementations. Demo implementations must be replaced for any real deployment: - **`inMemoryCounter`** — a process-local `counter.Counter` for sequence numbers; not durable. A real deployment uses a persistent implementation (e.g. `platform/extension/counter/mysql`). - **`fakeSourceControlFactory`** — seeds each queue with a deterministic single-commit history so ingest resolves a stable head URI (and re-ingesting the same queue exercises the dedup path). A real deployment supplies a VCS-backed `sourcecontrol.Factory`, which is also where a queue's promotion ref is resolved. The fake has no ref to move, so a promotion locally shows up only in the record consumer's logs. +- **`noopRepairExecutorFactory`** — accepts repairs without launching work and returns `noop/`. The fix controller stores that ID, consuming the build's one-repair allowance. Replacing this factory does not retry builds that already have a no-op association. The default no-op issue detector reports no failures, so the reference stack does not normally reach repair submission. ## Layout diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index 7c896fff6..27a199acd 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/service/stovepipe/server/BUILD.bazel @@ -29,6 +29,7 @@ go_library( "//stovepipe/controller/buildanalysis:go_default_library", "//stovepipe/controller/buildsignal:go_default_library", "//stovepipe/controller/dlq:go_default_library", + "//stovepipe/controller/fix:go_default_library", "//stovepipe/controller/issuetrack:go_default_library", "//stovepipe/controller/process:go_default_library", "//stovepipe/controller/record:go_default_library", @@ -38,6 +39,8 @@ go_library( "//stovepipe/extension/buildissue/noop:go_default_library", "//stovepipe/extension/buildrunner:go_default_library", "//stovepipe/extension/buildrunner/fake:go_default_library", + "//stovepipe/extension/fixer:go_default_library", + "//stovepipe/extension/fixer/noop:go_default_library", "//stovepipe/extension/projectstatus/noop:go_default_library", "//stovepipe/extension/queueconfig/default:go_default_library", "//stovepipe/extension/sourcecontrol:go_default_library", @@ -86,15 +89,21 @@ go_test( embed = [":go_default_library"], deps = [ "//api/base/hook:go_default_library", + "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/extension/messagequeue:go_default_library", "//stovepipe/controller/dlq:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/core/requestlog:go_default_library", + "//stovepipe/entity:go_default_library", "//stovepipe/extension/buildissue/noop:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "//stovepipe/extension/storage/mock:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", "@org_uber_go_zap//zaptest:go_default_library", ], ) diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 94d6d2167..c8a78c545 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -51,6 +51,7 @@ import ( "github.com/uber/submitqueue/stovepipe/controller/buildanalysis" "github.com/uber/submitqueue/stovepipe/controller/buildsignal" "github.com/uber/submitqueue/stovepipe/controller/dlq" + "github.com/uber/submitqueue/stovepipe/controller/fix" "github.com/uber/submitqueue/stovepipe/controller/issuetrack" "github.com/uber/submitqueue/stovepipe/controller/process" "github.com/uber/submitqueue/stovepipe/controller/record" @@ -60,6 +61,8 @@ import ( buildissuenoop "github.com/uber/submitqueue/stovepipe/extension/buildissue/noop" "github.com/uber/submitqueue/stovepipe/extension/buildrunner" buildrunnerfake "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake" + "github.com/uber/submitqueue/stovepipe/extension/fixer" + fixernoop "github.com/uber/submitqueue/stovepipe/extension/fixer/noop" projectstatusnoop "github.com/uber/submitqueue/stovepipe/extension/projectstatus/noop" queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default" "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" @@ -92,6 +95,12 @@ func (fakeBuildRunnerFactory) For(cfg buildrunner.Config) (buildrunner.BuildRunn return buildrunnerfake.New(cfg), nil } +type noopRepairExecutorFactory struct{} + +func (noopRepairExecutorFactory) For(fixer.Config) (fixer.RepairExecutor, error) { + return fixernoop.New(), nil +} + // hookResolver sends every event to the no-op hook. Which hooks an event goes to is host // policy, so the resolver lives here rather than in the extension package. A deployment // with real integrations swaps this for one that selects on the event's source and type. @@ -248,10 +257,11 @@ func run() error { // silently duplicated across controllers. sourceControl := fakeSourceControlFactory{} brf := fakeBuildRunnerFactory{} + repairExecutors := noopRepairExecutorFactory{} storageFty := storageFactory{backend: store} materializer := requestlog.NewMaterializer(scope) - primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl, brf, hookResolver{}, detectorFactory) + primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl, brf, hookResolver{}, detectorFactory, repairExecutors) if err != nil { return err } @@ -371,6 +381,7 @@ func registerPrimaryControllers( brf buildrunner.Factory, hooks hookext.Hooks, detectorFactory buildissue.DetectorFactory, + repairExecutors fixer.Factory, ) (int, error) { var count int @@ -414,6 +425,12 @@ func registerPrimaryControllers( } count++ + fixController := fix.NewController(logger, scope, store, repairExecutors, stovepipemq.TopicKeyFix, "stovepipe-fix") + if err := c.Register(fixController); err != nil { + return count, fmt.Errorf("failed to register fix controller: %w", err) + } + count++ + recordController := record.NewController(logger, scope, store, materializer, projectstatusnoop.New(), sourceControl, registry, stovepipemq.TopicKeyRecord, "stovepipe-record") if err := c.Register(recordController); err != nil { return count, fmt.Errorf("failed to register record controller: %w", err) @@ -580,9 +597,10 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-issuetrack-dlq"), }, { - Key: stovepipemq.TopicKeyFix, - Name: "fix", - Queue: q, + Key: stovepipemq.TopicKeyFix, + Name: "fix", + Queue: q, + Subscription: extqueue.ExtendedSubscriptionConfig(subscriberName, "stovepipe-fix"), }, { Key: stovepipemq.TopicKeyBuildAnalysis, diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go index aa1c10267..3a1907762 100644 --- a/service/stovepipe/server/main_test.go +++ b/service/stovepipe/server/main_test.go @@ -23,12 +23,18 @@ import ( "github.com/stretchr/testify/require" "github.com/uber-go/tally" basehook "github.com/uber/submitqueue/api/base/hook" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" "github.com/uber/submitqueue/stovepipe/controller/dlq" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/core/requestlog" + "github.com/uber/submitqueue/stovepipe/entity" buildissuenoop "github.com/uber/submitqueue/stovepipe/extension/buildissue/noop" + "github.com/uber/submitqueue/stovepipe/extension/storage" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" "go.uber.org/zap/zaptest" ) @@ -61,7 +67,7 @@ func registeredControllers(t *testing.T) (consumer.TopicRegistry, []consumer.Con deadLetter := &recordingConsumer{} _, err = registerPrimaryControllers(primary, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry, - fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{}, buildissuenoop.New()) + fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{}, buildissuenoop.New(), noopRepairExecutorFactory{}) require.NoError(t, err) _, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry) @@ -110,22 +116,86 @@ func TestIssueTrackingRegistration(t *testing.T) { assert.Equal(t, "_dlq", cfg.DLQ.TopicSuffix) } -func TestEveryPrimaryStageHasRecovery(t *testing.T) { +func TestValidationStagesHaveRecovery(t *testing.T) { _, primary, deadLetter := registeredControllers(t) recoveryKeys := topicKeys(deadLetter) for _, controller := range primary { + if controller.TopicKey() == stovepipemq.TopicKeyFix { + continue + } t.Run(controller.Name(), func(t *testing.T) { assert.Contains(t, recoveryKeys, dlq.TopicKey(controller.TopicKey())) }) } } -func TestFixHasPublicationRouteWithoutConsumption(t *testing.T) { - registry, primary, _ := registeredControllers(t) +func TestFixRegistration(t *testing.T) { + registry, primary, deadLetter := registeredControllers(t) name, ok := registry.TopicName(stovepipemq.TopicKeyFix) require.True(t, ok) assert.Equal(t, "fix", name) - assert.NotContains(t, topicKeys(primary), stovepipemq.TopicKeyFix) + assert.Contains(t, topicKeys(primary), stovepipemq.TopicKeyFix) + assert.NotContains(t, topicKeys(deadLetter), dlq.TopicKey(stovepipemq.TopicKeyFix)) +} + +func TestRegisteredFixControllerRecordsNoopRepairOncePerBuild(t *testing.T) { + const queueName = "validation/main" + const buildID = "build/1" + ctrl := gomock.NewController(t) + stores := storagemock.NewMockFactory(ctrl) + store := storagemock.NewMockStorage(ctrl) + observations := storagemock.NewMockBuildIssueStore(ctrl) + issues := storagemock.NewMockIssueStateStore(ctrl) + stores.EXPECT().For(storage.Config{QueueName: queueName}).Return(store, nil).AnyTimes() + store.EXPECT().GetBuildIssueStore().Return(observations).AnyTimes() + store.EXPECT().GetIssueStateStore().Return(issues).AnyTimes() + + observation := entity.BuildIssue{TaskID: "test", UnitID: "unit", Version: 1} + observations.EXPECT().List(gomock.Any(), buildID, gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, bounds storage.BuildIssueRange) ([]entity.BuildIssue, error) { + if bounds.After.TaskID != "" { + return nil, nil + } + return []entity.BuildIssue{observation}, nil + }).AnyTimes() + issues.EXPECT().Get(gomock.Any(), observation.TaskID, observation.UnitID).Return(entity.IssueState{ + TaskID: observation.TaskID, UnitID: observation.UnitID, Status: entity.IssueStatusOpen, + FirstSeenBuildID: buildID, LastProcessedBuildID: buildID, LastChangedBuildID: buildID, + ChangedAtMs: 1000, Version: 1, + }, nil) + observations.EXPECT().Update(gomock.Any(), buildID, gomock.Any(), int32(1), int32(2)).DoAndReturn( + func(_ context.Context, _ string, updated entity.BuildIssue, _, newVersion int32) error { + require.Equal(t, "noop/"+buildID, updated.FixID) + observation = updated + observation.Version = newVersion + return nil + }) + + registry, err := newTopicRegistry(nil, "subscriber") + require.NoError(t, err) + primary := &recordingConsumer{} + _, err = registerPrimaryControllers(primary, zaptest.NewLogger(t).Sugar(), tally.NoopScope, stores, + requestlog.NewMaterializer(tally.NoopScope), registry, fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, + hookResolver{}, buildissuenoop.New(), noopRepairExecutorFactory{}) + require.NoError(t, err) + var fixController consumer.Controller + for _, controller := range primary.controllers { + if controller.TopicKey() == stovepipemq.TopicKeyFix { + fixController = controller + break + } + } + require.NotNil(t, fixController) + + payload, err := stovepipemq.Marshal(&stovepipemq.FixRequest{Id: buildID, QueueName: queueName}) + require.NoError(t, err) + for _, pass := range []string{"pass/1", "pass/2"} { + message := entityqueue.NewMessage(buildID+"/"+pass, payload, buildID, nil) + message.Tenant = queueName + delivery := consumermock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(message) + require.NoError(t, fixController.Process(context.Background(), delivery)) + } } func TestHookStage(t *testing.T) { @@ -168,6 +238,7 @@ func TestPipelineSubscriptionRetryPolicies(t *testing.T) { {name: "buildsignal", topicKey: stovepipemq.TopicKeyBuildSignal, consumerGroup: "stovepipe-buildsignal", wantRetry: extendedRetry}, {name: "buildanalysis", topicKey: stovepipemq.TopicKeyBuildAnalysis, consumerGroup: "stovepipe-buildanalysis", wantRetry: extendedRetry}, {name: "issuetrack", topicKey: stovepipemq.TopicKeyIssueTrack, consumerGroup: "stovepipe-issuetrack", wantRetry: extendedRetry}, + {name: "fix", topicKey: stovepipemq.TopicKeyFix, consumerGroup: "stovepipe-fix", wantRetry: extendedRetry}, {name: "record", topicKey: stovepipemq.TopicKeyRecord, consumerGroup: "stovepipe-record", wantRetry: extendedRetry}, {name: "hook", topicKey: basehook.TopicKeyHook, consumerGroup: "stovepipe-hook", wantRetry: defaultRetry}, } diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md index 591a39099..68cbbf2cd 100644 --- a/stovepipe/controller/issuetrack/README.md +++ b/stovepipe/controller/issuetrack/README.md @@ -59,7 +59,7 @@ A notification can fail while helping an older pass. Recovery abandons that save Storage and history failures return for retry; hosts must pair recovery with an always-retryable, unlimited-retry DLQ consumer. Only recovery work repeats. Malformed message identities, unresolvable queues, and missing source records are logged and acknowledged. -The reference service registers queue-partitioned `issuetrack` and `issuetrack_dlq` subscriptions. Normal processing uses backend error classification and the extended bounded retry policy, while recovery uses the existing always-retryable DLQ consumer. The DLQ consumer starts before normal consumption and stops after it. Deploy recovery-aware controllers and provision the issue tables before enabling consumption; older controllers do not recognize the abandonment phases. Completed opening passes publish repair notifications; recovery does not. The fix publication route is registered, but fix consumption and executor configuration remain separate. +The reference service registers queue-partitioned `issuetrack` and `issuetrack_dlq` subscriptions. Normal processing uses backend error classification and the extended bounded retry policy, while recovery uses the existing always-retryable DLQ consumer. The DLQ consumer starts before normal consumption and stops after it. Deploy recovery-aware controllers and provision the issue tables before enabling consumption; older controllers do not recognize the abandonment phases. Completed opening passes publish repair notifications; recovery does not. The reference host also consumes repair notifications with a no-op executor; fix dead-letter recovery remains separate. ## Analysis handoff diff --git a/stovepipe/extension/fixer/noop/BUILD.bazel b/stovepipe/extension/fixer/noop/BUILD.bazel new file mode 100644 index 000000000..c384b8e89 --- /dev/null +++ b/stovepipe/extension/fixer/noop/BUILD.bazel @@ -0,0 +1,12 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["noop.go"], + importpath = "github.com/uber/submitqueue/stovepipe/extension/fixer/noop", + visibility = ["//visibility:public"], + deps = [ + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/fixer:go_default_library", + ], +) diff --git a/stovepipe/extension/fixer/noop/noop.go b/stovepipe/extension/fixer/noop/noop.go new file mode 100644 index 000000000..7ab458c96 --- /dev/null +++ b/stovepipe/extension/fixer/noop/noop.go @@ -0,0 +1,34 @@ +// 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 noop provides a repair executor that accepts submissions without launching work. +package noop + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/fixer" +) + +type executor struct{} + +// New returns a repair executor that accepts submissions without launching work. +func New() fixer.RepairExecutor { + return executor{} +} + +func (executor) Submit(_ context.Context, issue entity.OpenIssue) (string, error) { + return "noop/" + issue.FirstSeenBuildID, nil +}