From 5578a355ba15865d9fd09df0279af940205148a2 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Tue, 14 Jul 2026 15:46:21 +0000 Subject: [PATCH 1/3] feat(stovepipe): derive build strategy in process Pin each admitted request to the queue's last-green baseline, falling back safely to full builds when ancestry is unavailable. --- service/stovepipe/server/main.go | 10 +- stovepipe/controller/process/BUILD.bazel | 3 + stovepipe/controller/process/process.go | 49 +++++- stovepipe/controller/process/process_test.go | 170 ++++++++++++++++++- 4 files changed, 221 insertions(+), 11 deletions(-) diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 22c53a60..73035dfd 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -210,7 +210,15 @@ func run() error { ), ) - processController := process.NewController(logger.Sugar(), scope, store, queueconfigdefault.NewStore(), stovepipemq.TopicKeyProcess, "stovepipe-process") + processController := process.NewController( + logger.Sugar(), + scope, + store, + queueconfigdefault.NewStore(), + fakeSourceControlFactory{}, + stovepipemq.TopicKeyProcess, + "stovepipe-process", + ) if err := primaryConsumer.Register(processController); err != nil { return fmt.Errorf("failed to register process controller: %w", err) } diff --git a/stovepipe/controller/process/BUILD.bazel b/stovepipe/controller/process/BUILD.bazel index a856f647..b55513c1 100644 --- a/stovepipe/controller/process/BUILD.bazel +++ b/stovepipe/controller/process/BUILD.bazel @@ -12,6 +12,7 @@ go_library( "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/queueconfig:go_default_library", + "//stovepipe/extension/sourcecontrol:go_default_library", "//stovepipe/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_zap//:go_default_library", @@ -30,6 +31,8 @@ go_test( "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/queueconfig/default:go_default_library", + "//stovepipe/extension/sourcecontrol:go_default_library", + "//stovepipe/extension/sourcecontrol/mock:go_default_library", "//stovepipe/extension/storage:go_default_library", "//stovepipe/extension/storage/mock:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index dd80fdd1..c9496578 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -30,6 +30,7 @@ import ( stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/queueconfig" + "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/zap" ) @@ -42,6 +43,7 @@ type Controller struct { metricsScope tally.Scope store storage.Storage queueConfigs queueconfig.Store + sourceControl sourcecontrol.Factory topicKey consumer.TopicKey consumerGroup string } @@ -58,6 +60,7 @@ func NewController( scope tally.Scope, store storage.Storage, queueConfigs queueconfig.Store, + sourceControl sourcecontrol.Factory, topicKey consumer.TopicKey, consumerGroup string, ) *Controller { @@ -66,6 +69,7 @@ func NewController( metricsScope: scope.SubScope("process_controller"), store: store, queueConfigs: queueConfigs, + sourceControl: sourceControl, topicKey: topicKey, consumerGroup: consumerGroup, } @@ -186,8 +190,15 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request return nil } - err := c.claimBuildSlot(ctx, &queueRow) + strategy, baseURI, err := c.deriveBuildStrategy(ctx, queueRow, request) + if err != nil { + return err + } + + err = c.claimBuildSlot(ctx, &queueRow) if err == nil { + request.BuildStrategy = strategy + request.BaseURI = baseURI break } if !errors.Is(err, storage.ErrVersionMismatch) { @@ -201,10 +212,6 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request } } - // TODO(build-strategy): derive from queue last_green_uri + SourceControl.IsAncestor. - request.BuildStrategy = entity.BuildStrategyFull - request.BaseURI = "" - transitioned, err := c.markProcessing(ctx, &request) if err != nil { // Slot claimed but never admitted: release best-effort so the slot isn't leaked @@ -225,10 +232,42 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request "queue", request.Queue, "uri", request.URI, "build_strategy", string(request.BuildStrategy), + "base_uri", request.BaseURI, ) return nil } +// deriveBuildStrategy chooses the validation scope and baseline from the queue's last-known-good commit. +// The caller persists the returned values only after successfully claiming a build slot. +func (c *Controller) deriveBuildStrategy(ctx context.Context, queueRow entity.Queue, request entity.Request) (strategy entity.BuildStrategy, baseURI string, err error) { + if queueRow.LastGreenURI == "" { + return entity.BuildStrategyFull, "", nil + } + + sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue}) + if err != nil { + return entity.BuildStrategyUnknown, "", fmt.Errorf("ProcessController failed to resolve source control for queue %s: %w", request.Queue, err) + } + + isAncestor, err := sc.IsAncestor(ctx, queueRow.LastGreenURI, request.URI) + if err != nil { + if sourcecontrol.IsNotFound(err) { + c.logger.Warnw("last-green URI is not in request history; using full build", + "queue", request.Queue, + "last_green_uri", queueRow.LastGreenURI, + "request_uri", request.URI, + ) + return entity.BuildStrategyFull, "", nil + } + return entity.BuildStrategyUnknown, "", fmt.Errorf("ProcessController failed to check ancestry for queue %s: %w", request.Queue, err) + } + + if isAncestor { + return entity.BuildStrategyIncrementalSinceGreen, queueRow.LastGreenURI, nil + } + return entity.BuildStrategyFull, "", nil +} + // claimBuildSlot CAS-increments queue.in_flight_count by one. On version mismatch it // reloads queueRow and returns ErrVersionMismatch so the caller can retry. func (c *Controller) claimBuildSlot(ctx context.Context, queueRow *entity.Queue) error { diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index 610223a0..9bc9cd81 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -29,6 +29,8 @@ import ( stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default" + "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" + sourcecontrolmock "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/mock" "github.com/uber/submitqueue/stovepipe/extension/storage" storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" "go.uber.org/mock/gomock" @@ -43,23 +45,35 @@ const ( ) type processMocks struct { - reqStore *storagemock.MockRequestStore - queueStore *storagemock.MockQueueStore + reqStore *storagemock.MockRequestStore + queueStore *storagemock.MockQueueStore + sourceFactory *sourcecontrolmock.MockFactory + sourceControl *sourcecontrolmock.MockSourceControl } func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processMocks) { t.Helper() m := processMocks{ - reqStore: storagemock.NewMockRequestStore(ctrl), - queueStore: storagemock.NewMockQueueStore(ctrl), + reqStore: storagemock.NewMockRequestStore(ctrl), + queueStore: storagemock.NewMockQueueStore(ctrl), + sourceFactory: sourcecontrolmock.NewMockFactory(ctrl), + sourceControl: sourcecontrolmock.NewMockSourceControl(ctrl), } store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() - c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, queueconfigdefault.NewStore(), stovepipemq.TopicKeyProcess, "stovepipe-process") + c := NewController( + zap.NewNop().Sugar(), + tally.NewTestScope("test", nil), + store, + queueconfigdefault.NewStore(), + m.sourceFactory, + stovepipemq.TopicKeyProcess, + "stovepipe-process", + ) return c, m } @@ -103,6 +117,136 @@ func expectAdmit(m processMocks, id string) { m.reqStore.EXPECT().Update(gomock.Any(), updatedReq, int32(1), int32(2)).Return(nil) } +func TestDeriveBuildStrategy(t *testing.T) { + const lastGreenURI = "git://repo/monorepo/main/green" + + tests := []struct { + name string + queue entity.Queue + setup func(m processMocks) + wantStrategy entity.BuildStrategy + wantBaseURI string + wantErr bool + }{ + { + name: "cold start uses full build without source control", + queue: entity.Queue{Name: testQueue}, + wantStrategy: entity.BuildStrategyFull, + }, + { + name: "ancestor uses incremental build", + queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, + setup: func(m processMocks) { + m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(true, nil) + }, + wantStrategy: entity.BuildStrategyIncrementalSinceGreen, + wantBaseURI: lastGreenURI, + }, + { + name: "history rewrite uses full build", + queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, + setup: func(m processMocks) { + m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(false, nil) + }, + wantStrategy: entity.BuildStrategyFull, + }, + { + name: "unknown ancestry uses full build", + queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, + setup: func(m processMocks) { + m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(false, sourcecontrol.ErrNotFound) + }, + wantStrategy: entity.BuildStrategyFull, + }, + { + name: "factory error fails", + queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, + setup: func(m processMocks) { + m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(nil, errors.New("source control unavailable")) + }, + wantErr: true, + }, + { + name: "ancestry error fails", + queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, + setup: func(m processMocks) { + m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(false, errors.New("source control unavailable")) + }, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + if tt.setup != nil { + tt.setup(m) + } + + strategy, baseURI, err := c.deriveBuildStrategy(context.Background(), tt.queue, acceptedRequest(testID)) + + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.wantStrategy, strategy) + assert.Equal(t, tt.wantBaseURI, baseURI) + }) + } +} + +func TestProcessRederivesStrategyAfterQueueReload(t *testing.T) { + const ( + initialLastGreen = "git://repo/monorepo/main/green-old" + reloadedLastGreen = "git://repo/monorepo/main/green-new" + ) + + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + request := acceptedRequest(testID) + + initialQueue := entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + LastGreenURI: initialLastGreen, + Version: 1, + } + reloadedQueue := entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + LastGreenURI: reloadedLastGreen, + Version: 2, + } + claimedQueue := reloadedQueue + claimedQueue.InFlightCount = 1 + initialClaim := initialQueue + initialClaim.InFlightCount = 1 + + updatedRequest := request + updatedRequest.State = entity.RequestStateProcessing + updatedRequest.BuildStrategy = entity.BuildStrategyIncrementalSinceGreen + updatedRequest.BaseURI = reloadedLastGreen + + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(initialQueue, nil) + m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), initialLastGreen, testURI).Return(true, nil) + m.queueStore.EXPECT().Update(gomock.Any(), initialClaim, int32(1), int32(2)).Return(storage.ErrVersionMismatch) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(reloadedQueue, nil) + m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), reloadedLastGreen, testURI).Return(true, nil) + m.queueStore.EXPECT().Update(gomock.Any(), claimedQueue, int32(2), int32(3)).Return(nil) + m.reqStore.EXPECT().Update(gomock.Any(), updatedRequest, int32(1), int32(2)).Return(nil) + + require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID)))) +} + func TestProcess(t *testing.T) { tests := []struct { name string @@ -148,6 +292,22 @@ func TestProcess(t *testing.T) { expectAdmit(m, testID) }, }, + { + name: "source control failure does not claim a build slot", + wantErr: true, + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + LastGreenURI: "git://repo/monorepo/main/green", + Version: 1, + }, nil) + m.sourceFactory.EXPECT(). + For(sourcecontrol.Config{QueueName: testQueue}). + Return(nil, errors.New("source control unavailable")) + }, + }, { name: "accepted with empty latest pointer awaits ingest stamp", setup: func(m processMocks) { From 17bbcb8e80111e6129bd56df494fee984a1bdbf5 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Tue, 14 Jul 2026 18:22:09 +0000 Subject: [PATCH 2/3] fix(stovepipe): preserve build strategy on admit retries Reuse source-control resolution across queue CAS retries, record strategy outcomes, and keep the chosen strategy when a request CAS reloads an accepted row. --- stovepipe/controller/process/process.go | 49 +++++--- stovepipe/controller/process/process_test.go | 119 ++++++++++++++++--- 2 files changed, 137 insertions(+), 31 deletions(-) diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index c9496578..0c0bf35c 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -178,6 +178,11 @@ func (c *Controller) coalesce(ctx context.Context, request entity.Request, lates // re-runs coalesce-then-gate, so a slot is never spent on a now-stale head; a closed gate // defers (acks) rather than failing. func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request, queueRow entity.Queue, maxConcurrent int32) error { + var sc sourcecontrol.SourceControl + var strategy entity.BuildStrategy + var baseURI string + var err error + for { if queueRow.InFlightCount >= maxConcurrent { // TODO: re-enqueue the request via PublishAfter on the process topic with GateWaitDelayMs. @@ -190,15 +195,22 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request return nil } - strategy, baseURI, err := c.deriveBuildStrategy(ctx, queueRow, request) + if queueRow.LastGreenURI != "" && sc == nil { + var err error + sc, err = c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue}) + if err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1) + return fmt.Errorf("ProcessController failed to resolve source control for queue %s: %w", request.Queue, err) + } + } + + strategy, baseURI, err = c.deriveBuildStrategy(ctx, sc, queueRow, request) if err != nil { return err } err = c.claimBuildSlot(ctx, &queueRow) if err == nil { - request.BuildStrategy = strategy - request.BaseURI = baseURI break } if !errors.Is(err, storage.ErrVersionMismatch) { @@ -212,7 +224,7 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request } } - transitioned, err := c.markProcessing(ctx, &request) + transitioned, err := c.markProcessing(ctx, &request, strategy, baseURI) if err != nil { // Slot claimed but never admitted: release best-effort so the slot isn't leaked // (a redelivery would find the gate closed by its own claim and nothing decrements it). @@ -227,6 +239,9 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request // TODO(build-publish): publish BuildRequest to the build stage here. + metrics.NamedCounter(c.metricsScope, _opName, "admitted", 1, + metrics.NewTag("strategy", string(request.BuildStrategy)), + ) c.logger.Infow("admitted request to build", "request_id", request.ID, "queue", request.Queue, @@ -238,20 +253,18 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request } // deriveBuildStrategy chooses the validation scope and baseline from the queue's last-known-good commit. -// The caller persists the returned values only after successfully claiming a build slot. -func (c *Controller) deriveBuildStrategy(ctx context.Context, queueRow entity.Queue, request entity.Request) (strategy entity.BuildStrategy, baseURI string, err error) { +// The caller resolves source control once and persists the returned values only after successfully claiming a build slot. +func (c *Controller) deriveBuildStrategy(ctx context.Context, sc sourcecontrol.SourceControl, queueRow entity.Queue, request entity.Request) (strategy entity.BuildStrategy, baseURI string, err error) { if queueRow.LastGreenURI == "" { return entity.BuildStrategyFull, "", nil } - sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue}) - if err != nil { - return entity.BuildStrategyUnknown, "", fmt.Errorf("ProcessController failed to resolve source control for queue %s: %w", request.Queue, err) - } - isAncestor, err := sc.IsAncestor(ctx, queueRow.LastGreenURI, request.URI) if err != nil { if sourcecontrol.IsNotFound(err) { + metrics.NamedCounter(c.metricsScope, _opName, "strategy_fallbacks", 1, + metrics.NewTag("reason", "unknown_ancestry"), + ) c.logger.Warnw("last-green URI is not in request history; using full build", "queue", request.Queue, "last_green_uri", queueRow.LastGreenURI, @@ -259,6 +272,7 @@ func (c *Controller) deriveBuildStrategy(ctx context.Context, queueRow entity.Qu ) return entity.BuildStrategyFull, "", nil } + metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1) return entity.BuildStrategyUnknown, "", fmt.Errorf("ProcessController failed to check ancestry for queue %s: %w", request.Queue, err) } @@ -292,11 +306,12 @@ func (c *Controller) claimBuildSlot(ctx context.Context, queueRow *entity.Queue) return nil } -// markProcessing CAS-marks request accepted→processing, persisting BuildStrategy and BaseURI -// already set by the admit workflow. Retries on version conflicts. transitioned is true only -// when this call performed the CAS; false means a concurrent writer already advanced the -// request past accepted (a lost admit race), so the caller must release its claimed slot. -func (c *Controller) markProcessing(ctx context.Context, request *entity.Request) (transitioned bool, err error) { +// markProcessing CAS-marks request accepted→processing and persists the strategy chosen after the +// build slot was claimed. It reapplies the values after a request reload so an accepted concurrent +// update cannot discard them. transitioned is true only when this call performed the CAS; false +// means a concurrent writer already advanced the request past accepted, so the caller must release +// its claimed slot. +func (c *Controller) markProcessing(ctx context.Context, request *entity.Request, strategy entity.BuildStrategy, baseURI string) (transitioned bool, err error) { reqStore := c.store.GetRequestStore() for { @@ -306,6 +321,8 @@ func (c *Controller) markProcessing(ctx context.Context, request *entity.Request updated := *request updated.State = entity.RequestStateProcessing + updated.BuildStrategy = strategy + updated.BaseURI = baseURI newVersion := request.Version + 1 if err := reqStore.Update(ctx, updated, request.Version, newVersion); err != nil { if errors.Is(err, storage.ErrVersionMismatch) { diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index 9bc9cd81..4cc495e5 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -53,7 +53,11 @@ type processMocks struct { func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processMocks) { t.Helper() + return newControllerWithScope(t, ctrl, tally.NewTestScope("test", nil)) +} +func newControllerWithScope(t *testing.T, ctrl *gomock.Controller, scope tally.Scope) (*Controller, processMocks) { + t.Helper() m := processMocks{ reqStore: storagemock.NewMockRequestStore(ctrl), queueStore: storagemock.NewMockQueueStore(ctrl), @@ -67,7 +71,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processM c := NewController( zap.NewNop().Sugar(), - tally.NewTestScope("test", nil), + scope, store, queueconfigdefault.NewStore(), m.sourceFactory, @@ -137,7 +141,6 @@ func TestDeriveBuildStrategy(t *testing.T) { name: "ancestor uses incremental build", queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, setup: func(m processMocks) { - m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(true, nil) }, wantStrategy: entity.BuildStrategyIncrementalSinceGreen, @@ -147,7 +150,6 @@ func TestDeriveBuildStrategy(t *testing.T) { name: "history rewrite uses full build", queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, setup: func(m processMocks) { - m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(false, nil) }, wantStrategy: entity.BuildStrategyFull, @@ -156,24 +158,14 @@ func TestDeriveBuildStrategy(t *testing.T) { name: "unknown ancestry uses full build", queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, setup: func(m processMocks) { - m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(false, sourcecontrol.ErrNotFound) }, wantStrategy: entity.BuildStrategyFull, }, - { - name: "factory error fails", - queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, - setup: func(m processMocks) { - m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(nil, errors.New("source control unavailable")) - }, - wantErr: true, - }, { name: "ancestry error fails", queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, setup: func(m processMocks) { - m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(false, errors.New("source control unavailable")) }, wantErr: true, @@ -188,7 +180,11 @@ func TestDeriveBuildStrategy(t *testing.T) { tt.setup(m) } - strategy, baseURI, err := c.deriveBuildStrategy(context.Background(), tt.queue, acceptedRequest(testID)) + var sc sourcecontrol.SourceControl + if tt.queue.LastGreenURI != "" { + sc = m.sourceControl + } + strategy, baseURI, err := c.deriveBuildStrategy(context.Background(), sc, tt.queue, acceptedRequest(testID)) if tt.wantErr { require.Error(t, err) @@ -201,6 +197,74 @@ func TestDeriveBuildStrategy(t *testing.T) { } } +func TestDeriveBuildStrategyEmitsSourceControlMetrics(t *testing.T) { + const lastGreenURI = "git://repo/monorepo/main/green" + + tests := []struct { + name string + ancestryErr error + metricName string + metricTags string + }{ + { + name: "unknown ancestry records fallback", + ancestryErr: sourcecontrol.ErrNotFound, + metricName: "strategy_fallbacks", + metricTags: "reason=unknown_ancestry", + }, + { + name: "source control failure records error", + ancestryErr: errors.New("source control unavailable"), + metricName: "source_control_errors", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + scope := tally.NewTestScope("test", nil) + c, m := newControllerWithScope(t, ctrl, scope) + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(false, tt.ancestryErr) + + _, _, err := c.deriveBuildStrategy( + context.Background(), + m.sourceControl, + entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, + acceptedRequest(testID), + ) + + if sourcecontrol.IsNotFound(tt.ancestryErr) { + require.NoError(t, err) + } else { + require.Error(t, err) + } + counter, ok := scope.Snapshot().Counters()["test.process_controller.process."+tt.metricName+"+"+tt.metricTags] + require.True(t, ok) + assert.Equal(t, int64(1), counter.Value()) + }) + } +} + +func TestProcessEmitsAdmittedStrategyMetric(t *testing.T) { + ctrl := gomock.NewController(t) + scope := tally.NewTestScope("test", nil) + c, m := newControllerWithScope(t, ctrl, scope) + + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + Version: 1, + }, nil) + expectAdmit(m, testID) + + require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID)))) + + counter, ok := scope.Snapshot().Counters()["test.process_controller.process.admitted+strategy=full"] + require.True(t, ok) + assert.Equal(t, int64(1), counter.Value()) +} + func TestProcessRederivesStrategyAfterQueueReload(t *testing.T) { const ( initialLastGreen = "git://repo/monorepo/main/green-old" @@ -239,7 +303,6 @@ func TestProcessRederivesStrategyAfterQueueReload(t *testing.T) { m.sourceControl.EXPECT().IsAncestor(gomock.Any(), initialLastGreen, testURI).Return(true, nil) m.queueStore.EXPECT().Update(gomock.Any(), initialClaim, int32(1), int32(2)).Return(storage.ErrVersionMismatch) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(reloadedQueue, nil) - m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) m.sourceControl.EXPECT().IsAncestor(gomock.Any(), reloadedLastGreen, testURI).Return(true, nil) m.queueStore.EXPECT().Update(gomock.Any(), claimedQueue, int32(2), int32(3)).Return(nil) m.reqStore.EXPECT().Update(gomock.Any(), updatedRequest, int32(1), int32(2)).Return(nil) @@ -390,6 +453,32 @@ func TestProcess(t *testing.T) { m.reqStore.EXPECT().Update(gomock.Any(), superseded, int32(1), int32(2)).Return(nil) }, }, + { + name: "mark processing retry preserves derived strategy after accepted reload", + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, LatestRequestID: testID, Version: 1, + }, nil) + m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ + Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 1, + }, int32(1), int32(2)).Return(nil) + + firstAttempt := acceptedRequest(testID) + firstAttempt.State = entity.RequestStateProcessing + firstAttempt.BuildStrategy = entity.BuildStrategyFull + m.reqStore.EXPECT().Update(gomock.Any(), firstAttempt, int32(1), int32(2)).Return(storage.ErrVersionMismatch) + + reloaded := acceptedRequest(testID) + reloaded.Version = 2 + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(reloaded, nil) + + retry := reloaded + retry.State = entity.RequestStateProcessing + retry.BuildStrategy = entity.BuildStrategyFull + m.reqStore.EXPECT().Update(gomock.Any(), retry, int32(2), int32(3)).Return(nil) + }, + }, { name: "mark processing lost race releases slot and skips admit", setup: func(m processMocks) { From 77798186054a81c11ae1ead556db1981a4a0ce84 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Tue, 14 Jul 2026 22:13:35 +0000 Subject: [PATCH 3/3] fix(stovepipe): harden process strategy retries Distinguish source-control resolution from ancestry failures and pin retry behavior when queue or request rows reload. --- stovepipe/controller/process/process.go | 9 +- stovepipe/controller/process/process_test.go | 130 +++++++++++++------ 2 files changed, 97 insertions(+), 42 deletions(-) diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index 0c0bf35c..1dce57c1 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -196,10 +196,11 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request } if queueRow.LastGreenURI != "" && sc == nil { - var err error sc, err = c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue}) if err != nil { - metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1) + metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1, + metrics.NewTag("stage", "resolve"), + ) return fmt.Errorf("ProcessController failed to resolve source control for queue %s: %w", request.Queue, err) } } @@ -272,7 +273,9 @@ func (c *Controller) deriveBuildStrategy(ctx context.Context, sc sourcecontrol.S ) return entity.BuildStrategyFull, "", nil } - metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1) + metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1, + metrics.NewTag("stage", "ancestry"), + ) return entity.BuildStrategyUnknown, "", fmt.Errorf("ProcessController failed to check ancestry for queue %s: %w", request.Queue, err) } diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index 4cc495e5..13008712 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -216,6 +216,7 @@ func TestDeriveBuildStrategyEmitsSourceControlMetrics(t *testing.T) { name: "source control failure records error", ancestryErr: errors.New("source control unavailable"), metricName: "source_control_errors", + metricTags: "stage=ancestry", }, } @@ -265,49 +266,93 @@ func TestProcessEmitsAdmittedStrategyMetric(t *testing.T) { assert.Equal(t, int64(1), counter.Value()) } -func TestProcessRederivesStrategyAfterQueueReload(t *testing.T) { - const ( - initialLastGreen = "git://repo/monorepo/main/green-old" - reloadedLastGreen = "git://repo/monorepo/main/green-new" - ) +func TestProcessEmitsSourceControlResolutionMetric(t *testing.T) { + const lastGreenURI = "git://repo/monorepo/main/green" ctrl := gomock.NewController(t) - c, m := newController(t, ctrl) - request := acceptedRequest(testID) + scope := tally.NewTestScope("test", nil) + c, m := newControllerWithScope(t, ctrl, scope) - initialQueue := entity.Queue{ + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ Name: testQueue, LatestRequestID: testID, - LastGreenURI: initialLastGreen, + LastGreenURI: lastGreenURI, Version: 1, + }, nil) + m.sourceFactory.EXPECT(). + For(sourcecontrol.Config{QueueName: testQueue}). + Return(nil, errors.New("source control unavailable")) + + require.Error(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID)))) + + counter, ok := scope.Snapshot().Counters()["test.process_controller.process.source_control_errors+stage=resolve"] + require.True(t, ok) + assert.Equal(t, int64(1), counter.Value()) +} + +func TestProcessRederivesStrategyAfterQueueReload(t *testing.T) { + const ( + initialLastGreen = "git://repo/monorepo/main/green-old" + reloadedLastGreen = "git://repo/monorepo/main/green-new" + ) + + tests := []struct { + name string + initialLastGreen string + }{ + { + name: "rederives against changed baseline", + initialLastGreen: initialLastGreen, + }, + { + name: "resolves source control after baseline appears", + }, } - reloadedQueue := entity.Queue{ - Name: testQueue, - LatestRequestID: testID, - LastGreenURI: reloadedLastGreen, - Version: 2, - } - claimedQueue := reloadedQueue - claimedQueue.InFlightCount = 1 - initialClaim := initialQueue - initialClaim.InFlightCount = 1 - - updatedRequest := request - updatedRequest.State = entity.RequestStateProcessing - updatedRequest.BuildStrategy = entity.BuildStrategyIncrementalSinceGreen - updatedRequest.BaseURI = reloadedLastGreen - - m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) - m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(initialQueue, nil) - m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) - m.sourceControl.EXPECT().IsAncestor(gomock.Any(), initialLastGreen, testURI).Return(true, nil) - m.queueStore.EXPECT().Update(gomock.Any(), initialClaim, int32(1), int32(2)).Return(storage.ErrVersionMismatch) - m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(reloadedQueue, nil) - m.sourceControl.EXPECT().IsAncestor(gomock.Any(), reloadedLastGreen, testURI).Return(true, nil) - m.queueStore.EXPECT().Update(gomock.Any(), claimedQueue, int32(2), int32(3)).Return(nil) - m.reqStore.EXPECT().Update(gomock.Any(), updatedRequest, int32(1), int32(2)).Return(nil) - require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID)))) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + request := acceptedRequest(testID) + + initialQueue := entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + LastGreenURI: tt.initialLastGreen, + Version: 1, + } + reloadedQueue := entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + LastGreenURI: reloadedLastGreen, + Version: 2, + } + initialClaim := initialQueue + initialClaim.InFlightCount = 1 + claimedQueue := reloadedQueue + claimedQueue.InFlightCount = 1 + + updatedRequest := request + updatedRequest.State = entity.RequestStateProcessing + updatedRequest.BuildStrategy = entity.BuildStrategyIncrementalSinceGreen + updatedRequest.BaseURI = reloadedLastGreen + + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(initialQueue, nil) + m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) + if tt.initialLastGreen != "" { + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), tt.initialLastGreen, testURI).Return(true, nil) + } + m.queueStore.EXPECT().Update(gomock.Any(), initialClaim, int32(1), int32(2)).Return(storage.ErrVersionMismatch) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(reloadedQueue, nil) + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), reloadedLastGreen, testURI).Return(true, nil) + m.queueStore.EXPECT().Update(gomock.Any(), claimedQueue, int32(2), int32(3)).Return(nil) + m.reqStore.EXPECT().Update(gomock.Any(), updatedRequest, int32(1), int32(2)).Return(nil) + + require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID)))) + }) + } } func TestProcess(t *testing.T) { @@ -389,6 +434,7 @@ func TestProcess(t *testing.T) { Name: testQueue, LatestRequestID: testID, InFlightCount: 1, + LastGreenURI: "git://repo/monorepo/main/green", Version: 1, }, nil) }, @@ -456,17 +502,22 @@ func TestProcess(t *testing.T) { { name: "mark processing retry preserves derived strategy after accepted reload", setup: func(m processMocks) { + const lastGreenURI = "git://repo/monorepo/main/green" + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ - Name: testQueue, LatestRequestID: testID, Version: 1, + Name: testQueue, LatestRequestID: testID, LastGreenURI: lastGreenURI, Version: 1, }, nil) + m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(true, nil) m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 1, + Name: testQueue, LatestRequestID: testID, InFlightCount: 1, LastGreenURI: lastGreenURI, Version: 1, }, int32(1), int32(2)).Return(nil) firstAttempt := acceptedRequest(testID) firstAttempt.State = entity.RequestStateProcessing - firstAttempt.BuildStrategy = entity.BuildStrategyFull + firstAttempt.BuildStrategy = entity.BuildStrategyIncrementalSinceGreen + firstAttempt.BaseURI = lastGreenURI m.reqStore.EXPECT().Update(gomock.Any(), firstAttempt, int32(1), int32(2)).Return(storage.ErrVersionMismatch) reloaded := acceptedRequest(testID) @@ -475,7 +526,8 @@ func TestProcess(t *testing.T) { retry := reloaded retry.State = entity.RequestStateProcessing - retry.BuildStrategy = entity.BuildStrategyFull + retry.BuildStrategy = entity.BuildStrategyIncrementalSinceGreen + retry.BaseURI = lastGreenURI m.reqStore.EXPECT().Update(gomock.Any(), retry, int32(2), int32(3)).Return(nil) }, },