[draft/wip] feat(stovepipe): admit latest head and publish to build

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,16 +107,16 @@ 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 {
 		return err
 	}
 
 	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,106 @@ 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 {
+	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
+	}
+
+	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 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()
+
+	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
+			return storage.ErrVersionMismatch
+		}
+		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,
+// 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)
+		}
+		updated.Version = newVersion
+		*request = updated
+		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()
 

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,26 +119,27 @@ 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,
 				}, nil)
 			},
 		},
 		{
-			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{
 					Name:            testQueue,
 					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{