fix(stovepipe): Speed up E2E polling

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`).

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:

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)
 	}

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)

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,16 +70,22 @@ 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,
 		materializer:    materializer,
 		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",

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
 	}

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,