From 6e72e83c610aa0499323fe9dac67c97a31b482bd Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Thu, 12 Nov 2026 19:09:18 +0000 Subject: [PATCH] fix(stovepipe): Speed up E2E polling ## Summary **What**: - Use half-second analysis polling in Stovepipe end-to-end tests while preserving normal deployment defaults. **Why**: - Prevent fake-build scenarios from exhausting the suite's deadline on production-sized polling waits. Continue exercising the same pipeline and persistence behavior without increasing the timeout. ## Test Plan - [x] Add unit tests. ## Revert Plan - Revert this PR. No schema or persistent data changes are involved. ## Issue Links - [CODEM-632](https://linear.app/uber/issue/CODEM-632/projects-issues) --- service/stovepipe/README.md | 4 + service/stovepipe/docker-compose.yml | 1 + service/stovepipe/server/main.go | 19 ++++- service/stovepipe/server/main_test.go | 83 +++++++++++++++++++ .../controller/buildanalysis/buildanalysis.go | 21 ++++- stovepipe/controller/issuetrack/issuetrack.go | 35 ++++++-- test/e2e/stovepipe/suite_test.go | 1 + 7 files changed, 155 insertions(+), 9 deletions(-) diff --git a/service/stovepipe/README.md b/service/stovepipe/README.md index 3aace589e..6b06968be 100644 --- a/service/stovepipe/README.md +++ b/service/stovepipe/README.md @@ -22,6 +22,10 @@ Stovepipe therefore needs two MySQL databases: a **storage** database (the `queu ## Wiring notes +`ANALYSIS_POLL_DELAY_MS` optionally overrides build-analysis polling and issue +tracking's wait for analysis. When unset, the controllers retain their respective +two-minute and 30-second defaults. The E2E host uses 500 milliseconds. + `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`). diff --git a/service/stovepipe/docker-compose.yml b/service/stovepipe/docker-compose.yml index 035f5ea0f..d5c2fb429 100644 --- a/service/stovepipe/docker-compose.yml +++ b/service/stovepipe/docker-compose.yml @@ -71,6 +71,7 @@ services: # Level for the queue's own logs; info by default so its per-message # chatter does not bury the rest of the service at debug. - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} + - ANALYSIS_POLL_DELAY_MS=${ANALYSIS_POLL_DELAY_MS:-} - MQ_TENANTS=${MQ_TENANTS:-monorepo/main,monorepo/release,monorepo/slow?buildrunner-fake=build-slow} - HOSTNAME=stovepipe-dev depends_on: diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 267145772..cfd5c5bd1 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -22,6 +22,7 @@ import ( "net" "os" "os/signal" + "strconv" "sync" "syscall" "time" @@ -390,6 +391,20 @@ func registerPrimaryControllers( detectorFactory buildissue.DetectorFactory, repairExecutors fixer.Factory, ) (int, error) { + var buildAnalysisOptions []buildanalysis.Option + var issueTrackingOptions []issuetrack.Option + if value := os.Getenv("ANALYSIS_POLL_DELAY_MS"); value != "" { + delayMs, err := strconv.ParseInt(value, 10, 64) + if err != nil { + return 0, fmt.Errorf("parse ANALYSIS_POLL_DELAY_MS: %w", err) + } + if delayMs <= 0 { + return 0, fmt.Errorf("ANALYSIS_POLL_DELAY_MS must be positive") + } + buildAnalysisOptions = append(buildAnalysisOptions, buildanalysis.WithPollDelayMs(delayMs)) + issueTrackingOptions = append(issueTrackingOptions, issuetrack.WithAnalysisWaitMs(delayMs)) + } + var count int processController := process.NewController( @@ -420,13 +435,13 @@ func registerPrimaryControllers( } count++ - analysisController := buildanalysis.NewController(logger, scope, store, materializer, detectorFactory, stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis") + analysisController := buildanalysis.NewController(logger, scope, store, materializer, detectorFactory, stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis", buildAnalysisOptions...) if err := c.Register(analysisController); err != nil { return count, fmt.Errorf("failed to register buildanalysis controller: %w", err) } count++ - issueTrackController := issuetrack.NewController(logger, scope, store, materializer, registry, stovepipemq.TopicKeyIssueTrack, "stovepipe-issuetrack") + issueTrackController := issuetrack.NewController(logger, scope, store, materializer, registry, stovepipemq.TopicKeyIssueTrack, "stovepipe-issuetrack", issueTrackingOptions...) if err := c.Register(issueTrackController); err != nil { return count, fmt.Errorf("failed to register issuetrack controller: %w", err) } diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go index 57a11b331..6c3e8983c 100644 --- a/service/stovepipe/server/main_test.go +++ b/service/stovepipe/server/main_test.go @@ -98,6 +98,89 @@ func TestEveryRegisteredControllerResolvesInTheTopicRegistry(t *testing.T) { } } +func TestAnalysisPollingDelays(t *testing.T) { + for _, tt := range []struct { + name string + value string + analysisWait int64 + trackingWait int64 + wantErr bool + }{ + {name: "production defaults", analysisWait: 120000, trackingWait: 30000}, + {name: "E2E override", value: "500", analysisWait: 500, trackingWait: 500}, + {name: "defaults after override", analysisWait: 120000, trackingWait: 30000}, + {name: "zero", value: "0", wantErr: true}, + {name: "negative", value: "-1", wantErr: true}, + {name: "not an integer", value: "invalid", wantErr: true}, + } { + t.Run(tt.name, func(t *testing.T) { + t.Setenv("ANALYSIS_POLL_DELAY_MS", tt.value) + ctrl := gomock.NewController(t) + stores := storagemock.NewMockFactory(ctrl) + store := storagemock.NewMockStorage(ctrl) + 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{}) + if tt.wantErr { + require.ErrorContains(t, err, "ANALYSIS_POLL_DELAY_MS") + require.Empty(t, primary.controllers) + return + } + require.NoError(t, err) + + const queue = "validation/main" + requests := storagemock.NewMockRequestStore(ctrl) + builds := storagemock.NewMockBuildStore(ctrl) + observations := storagemock.NewMockBuildIssueStore(ctrl) + tracking := storagemock.NewMockIssueTrackingStore(ctrl) + history := storagemock.NewMockRequestLogStore(ctrl) + stores.EXPECT().For(storage.Config{QueueName: queue}).Return(store, nil).AnyTimes() + store.EXPECT().GetRequestStore().Return(requests).AnyTimes() + store.EXPECT().GetBuildStore().Return(builds).AnyTimes() + store.EXPECT().GetBuildIssueStore().Return(observations).AnyTimes() + store.EXPECT().GetIssueTrackingStore().Return(tracking).AnyTimes() + store.EXPECT().GetRequestLogStore().Return(history).AnyTimes() + requests.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{ + ID: "1", Queue: queue, State: entity.RequestStateProcessing, + }, nil).AnyTimes() + builds.EXPECT().Get(gomock.Any(), "build").Return(entity.Build{ + ID: "build", RequestID: "1", Status: entity.BuildStatusRunning, + }, nil).AnyTimes() + tracking.EXPECT().Get(gomock.Any()).Return(entity.IssueTrackingState{}, storage.ErrNotFound) + history.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(entity.RequestLog{}, storage.ErrNotFound).AnyTimes() + + checked := 0 + for _, controller := range primary.controllers { + var msg entityqueue.Message + var delayMs int64 + switch controller.TopicKey() { + case stovepipemq.TopicKeyBuildAnalysis: + payload, err := stovepipemq.Marshal(&stovepipemq.BuildAnalysis{Id: "build", QueueName: queue}) + require.NoError(t, err) + msg = entityqueue.Message{ID: "build/started", Tenant: queue, PartitionKey: "build", Payload: payload} + delayMs = tt.analysisWait + case stovepipemq.TopicKeyIssueTrack: + payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{RequestId: "1", QueueName: queue}) + require.NoError(t, err) + msg = entityqueue.Message{ID: "1/issues", Tenant: queue, PartitionKey: queue, Payload: payload} + delayMs = tt.trackingWait + default: + continue + } + delivery := consumermock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(msg) + delivery.EXPECT().Hold(delayMs) + require.NoError(t, controller.Process(context.Background(), delivery)) + checked++ + } + require.Equal(t, 2, checked) + }) + } +} + func TestIssueTrackingRegistration(t *testing.T) { registry, primary, deadLetter := registeredControllers(t) name, ok := registry.TopicName(stovepipemq.TopicKeyIssueTrack) diff --git a/stovepipe/controller/buildanalysis/buildanalysis.go b/stovepipe/controller/buildanalysis/buildanalysis.go index eaa448ad3..02b4188d8 100644 --- a/stovepipe/controller/buildanalysis/buildanalysis.go +++ b/stovepipe/controller/buildanalysis/buildanalysis.go @@ -39,6 +39,16 @@ import ( // Two minutes limits provider traffic during long-running builds. const PollDelayMs int64 = 120000 +// Option configures a build-analysis controller. +type Option func(*Controller) + +// WithPollDelayMs sets the delay between successful analysis passes, in milliseconds. +func WithPollDelayMs(delayMs int64) Option { + return func(c *Controller) { + c.pollDelayMs = delayMs + } +} + // Controller accumulates build issues while builds run and after completion. // Terminal analysis must persist all build issues and its completion event before acknowledging. type Controller struct { @@ -49,6 +59,7 @@ type Controller struct { detectorFactory buildissue.DetectorFactory topicKey consumer.TopicKey consumerGroup string + pollDelayMs int64 } func NewController( @@ -59,8 +70,9 @@ func NewController( detectorFactory buildissue.DetectorFactory, topicKey consumer.TopicKey, consumerGroup string, + options ...Option, ) *Controller { - return &Controller{ + controller := &Controller{ logger: logger.Named("buildanalysis_controller"), metricsScope: scope.SubScope("buildanalysis_controller"), stores: stores, @@ -68,7 +80,12 @@ func NewController( detectorFactory: detectorFactory, topicKey: topicKey, consumerGroup: consumerGroup, + pollDelayMs: PollDelayMs, + } + for _, option := range options { + option(controller) } + return controller } // Process persists detected issues and holds while further analysis is needed. @@ -176,7 +193,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r } if hold { - delivery.Hold(PollDelayMs) + delivery.Hold(c.pollDelayMs) } c.logger.Debugw("build analysis pass complete", diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go index 3a9fe872e..41336df78 100644 --- a/stovepipe/controller/issuetrack/issuetrack.go +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -47,6 +47,16 @@ const ( type issueKey struct{ task, unit string } +// Option configures an issue-tracking controller. +type Option func(*Controller) + +// WithAnalysisWaitMs sets the delay before rechecking build analysis, in milliseconds. +func WithAnalysisWaitMs(delayMs int64) Option { + return func(c *Controller) { + c.analysisWaitMs = delayMs + } +} + // Controller retains issue transitions and their queue-scoped progress. type Controller struct { logger *zap.SugaredLogger @@ -58,10 +68,20 @@ type Controller struct { topicKey consumer.TopicKey consumerGroup string nowMs func() int64 + analysisWaitMs int64 } -func NewController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, materializer requestlog.Materializer, registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string) *Controller { - return &Controller{ +func NewController( + logger *zap.SugaredLogger, + scope tally.Scope, + stores storage.Factory, + materializer requestlog.Materializer, + registry consumer.TopicRegistry, + topicKey consumer.TopicKey, + consumerGroup string, + options ...Option, +) *Controller { + controller := &Controller{ logger: logger.Named("issuetrack_controller"), metricsScope: scope.SubScope("issuetrack_controller"), stores: stores, @@ -71,7 +91,12 @@ func NewController(logger *zap.SugaredLogger, scope tally.Scope, stores storage. topicKey: topicKey, consumerGroup: consumerGroup, nowMs: func() int64 { return time.Now().UnixMilli() }, + analysisWaitMs: _analysisWaitMs, + } + for _, option := range options { + option(controller) } + return controller } func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { @@ -145,7 +170,7 @@ func (c *Controller) trackRequest(ctx context.Context, delivery consumer.Deliver } default: if order > 0 { - delivery.Hold(_analysisWaitMs) + delivery.Hold(c.analysisWaitMs) return nil } } @@ -160,7 +185,7 @@ func (c *Controller) trackRequest(ctx context.Context, delivery consumer.Deliver return nil } if !request.State.IsTerminal() { - delivery.Hold(_analysisWaitMs) + delivery.Hold(c.analysisWaitMs) return nil } } @@ -246,7 +271,7 @@ func (c *Controller) trackRequest(ctx context.Context, delivery consumer.Deliver case entity.IssueTrackingPhaseRefreshing, entity.IssueTrackingPhaseFinalizing: delivery.Hold(0) case entity.IssueTrackingPhaseRefreshed: - delivery.Hold(_analysisWaitMs) + delivery.Hold(c.analysisWaitMs) } return nil } diff --git a/test/e2e/stovepipe/suite_test.go b/test/e2e/stovepipe/suite_test.go index c2b54d0c8..975f42c81 100644 --- a/test/e2e/stovepipe/suite_test.go +++ b/test/e2e/stovepipe/suite_test.go @@ -62,6 +62,7 @@ func (s *StovepipeE2ESuite) SetupSuite() { t := s.T() s.ctx = context.Background() s.log = testutil.NewTestLogger(t) + t.Setenv("ANALYSIS_POLL_DELAY_MS", "500") t.Setenv("MQ_TENANTS", strings.Join([]string{ "monorepo/main", "monorepo/release", "monorepo/slow?buildrunner-fake=build-slow", listPaginationQueue, listTimeBoundsQueue, listPrimaryQueue, listOtherQueue, listValidationQueue, queueStateTestQueue,