From 44606ba42499176d93f8eda3c5d99883bd027d0c Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 10 Jul 2026 15:19:42 +0000 Subject: [PATCH 1/2] feat(stovepipe): admit latest head through processing transition MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Increment in_flight_count and transition accepted→processing with a cold-start full build strategy when the concurrency gate is open. Build queue publish lands in a follow-up PR. --- stovepipe/controller/process/process.go | 107 +++++++++++++++++-- stovepipe/controller/process/process_test.go | 26 ++++- 2 files changed, 117 insertions(+), 16 deletions(-) diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index 0ec2cc0b..94f962c3 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -14,8 +14,8 @@ // Package process holds the process-stage queue controller. It consumes request // ids from ingest, reloads the Request from storage, coalesces older heads, and -// (in later changes) gates concurrency, decides build strategy, and admits -// winners to build. +// admits the latest head when a build slot is open. Build queue publish lands in +// a follow-up PR. package process import ( @@ -35,7 +35,8 @@ import ( ) // Controller consumes ProcessRequest messages from the process stage, reloads the -// referenced Request from storage, and coalesces older heads. Implements consumer.Controller. +// referenced Request from storage, coalesces older heads, and admits the latest when +// a slot is open. Implements consumer.Controller. type Controller struct { logger *zap.SugaredLogger metricsScope tally.Scope @@ -67,8 +68,8 @@ func NewController( } } -// Process reloads the request referenced by the delivery and coalesces older heads. -// Returns nil to ack (success) or an error to nack (retry). +// Process reloads the request referenced by the delivery, coalesces older heads, +// and admits the latest when a slot is open. Returns nil to ack (success) or an error to nack (retry). func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { const opName = "process" @@ -92,6 +93,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r switch request.State { case entity.RequestStateSuperseded, entity.RequestStateProcessing: + // Processing republish to build lands in a follow-up PR. return nil case entity.RequestStateAccepted: return c.processAccepted(ctx, request) @@ -105,8 +107,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r } } -// processAccepted coalesces older heads against queue.latest_request_id, then resolves -// per-queue config for the concurrency gate. Admit lands in a follow-up PR. +// processAccepted coalesces older heads against queue.latest_request_id, then admits +// the latest head when a build slot is available. func (c *Controller) processAccepted(ctx context.Context, request entity.Request) error { queueRow, err := c.loadQueue(ctx, request.Queue) if err != nil { @@ -114,7 +116,7 @@ func (c *Controller) processAccepted(ctx context.Context, request entity.Request } if queueRow.LatestRequestID == "" { - c.logger.Infow("latest head awaiting admit", + c.logger.Infow("latest head awaiting ingest pointer", "request_id", request.ID, "queue", request.Queue, "uri", request.URI, @@ -152,15 +154,98 @@ func (c *Controller) processAccepted(ctx context.Context, request entity.Request return nil } - c.logger.Infow("latest head awaiting admit", + return c.admitRequest(ctx, request, queueRow, cfg.MaxConcurrent) +} + +// admitRequest admits the latest head: cold-start full build, increment in_flight_count, +// and transition accepted→processing. +func (c *Controller) admitRequest(ctx context.Context, request entity.Request, queueRow entity.Queue, maxConcurrent int32) error { + if err := c.incrementInFlightCount(ctx, &queueRow, maxConcurrent); err != nil { + return err + } + + strategy := entity.BuildStrategyFull + if err := c.transitionToProcessing(ctx, &request, strategy, ""); err != nil { + return err + } + + // TODO(build-publish): publish BuildRequest to the build stage here. + + c.logger.Infow("admitted request", "request_id", request.ID, "queue", request.Queue, - "uri", request.URI, + "build_strategy", string(strategy), ) return nil } -// supersedeRequest CAS-marks request accepted→superseded, retrying on version conflicts. +// incrementInFlightCount CAS-increments queue.in_flight_count, retrying on version conflicts. +func (c *Controller) incrementInFlightCount(ctx context.Context, queueRow *entity.Queue, maxConcurrent int32) error { + queueStore := c.store.GetQueueStore() + + for { + if queueRow.InFlightCount >= maxConcurrent { + return fmt.Errorf("ProcessController gate closed for queue %s", queueRow.Name) + } + + updated := *queueRow + updated.InFlightCount = queueRow.InFlightCount + 1 + newVersion := queueRow.Version + 1 + if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + got, getErr := queueStore.Get(ctx, queueRow.Name) + if getErr != nil { + return fmt.Errorf("ProcessController failed to reload queue %s after version mismatch: %w", queueRow.Name, getErr) + } + *queueRow = got + continue + } + return fmt.Errorf("ProcessController failed to increment in_flight_count for queue %s: %w", queueRow.Name, err) + } + *queueRow = updated + queueRow.Version = newVersion + return nil + } +} + +// transitionToProcessing CAS-marks request accepted→processing with strategy fields, +// retrying on version conflicts. +func (c *Controller) transitionToProcessing( + ctx context.Context, + request *entity.Request, + strategy entity.BuildStrategy, + baseURI string, +) error { + reqStore := c.store.GetRequestStore() + + for { + if request.State != entity.RequestStateAccepted { + return nil + } + + 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) { + got, getErr := reqStore.Get(ctx, request.ID) + if getErr != nil { + return fmt.Errorf("ProcessController failed to reload request %s after version mismatch: %w", request.ID, getErr) + } + *request = got + continue + } + return fmt.Errorf("ProcessController failed to transition request %s to processing: %w", request.ID, err) + } + *request = updated + request.Version = newVersion + return nil + } +} + +// supersedeRequest transitions a request from accepted to superseded, retrying on version conflicts. func (c *Controller) supersedeRequest(ctx context.Context, request entity.Request) error { reqStore := c.store.GetRequestStore() diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index 2fd124e6..ab6d016f 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -25,7 +25,7 @@ import ( entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" - queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default" @@ -64,7 +64,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processM func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) consumer.Delivery { t.Helper() - d := queuemock.NewMockDelivery(ctrl) + d := mqmock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(entityqueue.NewMessage(testID, payload, testQueue, nil)).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() return d @@ -87,6 +87,21 @@ func acceptedRequest(id string) entity.Request { } } +func expectAdmit(m processMocks, id string) { + updatedQueue := entity.Queue{ + Name: testQueue, + LatestRequestID: id, + InFlightCount: 1, + Version: 1, + } + m.queueStore.EXPECT().Update(gomock.Any(), updatedQueue, int32(1), int32(2)).Return(nil) + + updatedReq := acceptedRequest(id) + updatedReq.State = entity.RequestStateProcessing + updatedReq.BuildStrategy = entity.BuildStrategyFull + m.reqStore.EXPECT().Update(gomock.Any(), updatedReq, int32(1), int32(2)).Return(nil) +} + func TestProcess(t *testing.T) { tests := []struct { name string @@ -104,7 +119,7 @@ func TestProcess(t *testing.T) { }, }, { - name: "processing is no-op until republish lands", + name: "processing is no-op until build publish lands", setup: func(m processMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{ ID: testID, Queue: testQueue, State: entity.RequestStateProcessing, Version: 2, @@ -112,7 +127,7 @@ func TestProcess(t *testing.T) { }, }, { - name: "latest accepted head awaits admit", + name: "latest accepted head is admitted", 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{ @@ -120,10 +135,11 @@ func TestProcess(t *testing.T) { LatestRequestID: testID, Version: 1, }, nil) + expectAdmit(m, testID) }, }, { - name: "accepted with empty latest pointer awaits admit", + name: "accepted with empty latest pointer awaits ingest stamp", 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{ From 3504ac056b031cb61743c0cb79dee7efa84d230c Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 10 Jul 2026 20:35:58 +0000 Subject: [PATCH 2/2] refactor(stovepipe): centralize admit gate check in admitRequest Move max-concurrent enforcement into admitRequest's retry loop so incrementInFlightCount is a single CAS increment that reloads on version mismatch. --- stovepipe/controller/process/process.go | 56 ++++++++++++++----------- 1 file changed, 32 insertions(+), 24 deletions(-) diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index 94f962c3..2cb35707 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -160,7 +160,20 @@ func (c *Controller) processAccepted(ctx context.Context, request entity.Request // admitRequest admits the latest head: cold-start full build, increment in_flight_count, // and transition accepted→processing. func (c *Controller) admitRequest(ctx context.Context, request entity.Request, queueRow entity.Queue, maxConcurrent int32) error { - if err := c.incrementInFlightCount(ctx, &queueRow, maxConcurrent); err != nil { + for { + // Re-check after each reload: another admission may have filled the slot since + // processAccepted's read or since the last failed CAS attempt. + if queueRow.InFlightCount >= maxConcurrent { + return fmt.Errorf("ProcessController gate closed for queue %s", queueRow.Name) + } + + err := c.incrementInFlightCount(ctx, &queueRow) + if err == nil { + break + } + if errors.Is(err, storage.ErrVersionMismatch) { + continue + } return err } @@ -179,33 +192,28 @@ func (c *Controller) admitRequest(ctx context.Context, request entity.Request, q return nil } -// incrementInFlightCount CAS-increments queue.in_flight_count, retrying on version conflicts. -func (c *Controller) incrementInFlightCount(ctx context.Context, queueRow *entity.Queue, maxConcurrent int32) error { +// incrementInFlightCount 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) incrementInFlightCount(ctx context.Context, queueRow *entity.Queue) error { queueStore := c.store.GetQueueStore() - for { - if queueRow.InFlightCount >= maxConcurrent { - return fmt.Errorf("ProcessController gate closed for queue %s", queueRow.Name) - } - - updated := *queueRow - updated.InFlightCount = queueRow.InFlightCount + 1 - newVersion := queueRow.Version + 1 - if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil { - if errors.Is(err, storage.ErrVersionMismatch) { - got, getErr := queueStore.Get(ctx, queueRow.Name) - if getErr != nil { - return fmt.Errorf("ProcessController failed to reload queue %s after version mismatch: %w", queueRow.Name, getErr) - } - *queueRow = got - continue + updated := *queueRow + updated.InFlightCount = queueRow.InFlightCount + 1 + newVersion := queueRow.Version + 1 + if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + got, getErr := queueStore.Get(ctx, queueRow.Name) + if getErr != nil { + return fmt.Errorf("ProcessController failed to reload queue %s after version mismatch: %w", queueRow.Name, getErr) } - return fmt.Errorf("ProcessController failed to increment in_flight_count for queue %s: %w", queueRow.Name, err) + *queueRow = got + return storage.ErrVersionMismatch } - *queueRow = updated - queueRow.Version = newVersion - return nil + return fmt.Errorf("ProcessController failed to increment in_flight_count for queue %s: %w", queueRow.Name, err) } + updated.Version = newVersion + *queueRow = updated + return nil } // transitionToProcessing CAS-marks request accepted→processing with strategy fields, @@ -239,8 +247,8 @@ func (c *Controller) transitionToProcessing( } return fmt.Errorf("ProcessController failed to transition request %s to processing: %w", request.ID, err) } + updated.Version = newVersion *request = updated - request.Version = newVersion return nil } }