feat(stovepipe): reschedule process when concurrency gate is closed

Ack the current delivery and PublishAfter the same ProcessRequest when
the latest head cannot claim a build slot, so gate waits do not burn
MaxAttempts or block the partition.

Author: mnoah1

stovepipe/controller/process/process.go

@@ -22,6 +22,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/uber-go/tally"
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
@@ -153,7 +154,7 @@ func (c *Controller) processAccepted(ctx context.Context, request entity.Request
 		return fmt.Errorf("ProcessController failed to load queue config for %s: %w", request.Queue, err)
 	}
 
-	return c.admitLatestHead(ctx, request, queueRow, cfg.MaxConcurrent)
+	return c.admitLatestHead(ctx, request, queueRow, cfg)
 }
 
 // coalesce supersedes request when a newer head exists (RFC process step 5), returning
@@ -183,23 +184,16 @@ func (c *Controller) coalesce(ctx context.Context, request entity.Request, lates
 // admitLatestHead runs the gate-then-admit workflow for the latest head: claim a build
 // slot, mark the request processing, and publish it to build. Every queue-row reload
 // 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 {
+// defers by rescheduling the request (ack after re-enqueue) rather than failing.
+func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request, queueRow entity.Queue, cfg entity.QueueConfig) 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.
-			c.logger.Infow("latest head awaiting build slot",
-				"request_id", request.ID,
-				"queue", request.Queue,
-				"uri", request.URI,
-				"in_flight_count", queueRow.InFlightCount,
-			)
-			return nil
+		if queueRow.InFlightCount >= cfg.MaxConcurrent {
+			return c.rescheduleProcess(ctx, request, queueRow.InFlightCount, cfg.GateWaitDelayMs)
 		}
 
 		if queueRow.LastGreenURI != "" && sc == nil {
@@ -418,6 +412,47 @@ func (c *Controller) supersedeRequest(ctx context.Context, request entity.Reques
 	}
 }
 
+// rescheduleProcess re-enqueues the same ProcessRequest after a delay so the gate can be
+// re-checked without burning MaxAttempts. delayMs must be positive.
+func (c *Controller) rescheduleProcess(ctx context.Context, request entity.Request, inFlightCount int32, delayMs int64) error {
+	if delayMs <= 0 {
+		metrics.NamedCounter(c.metricsScope, _opName, "config_errors", 1)
+		return fmt.Errorf("ProcessController requires a positive gate wait delay for queue %s, got %dms", request.Queue, delayMs)
+	}
+
+	payload, err := stovepipemq.Marshal(&stovepipemq.ProcessRequest{Id: request.ID})
+	if err != nil {
+		return fmt.Errorf("ProcessController failed to serialize process request %s: %w", request.ID, err)
+	}
+
+	// Suffix the message id with the publish time so the reschedule can't collide with
+	// the in-flight delivery's still-present message-store row.
+	msgID := fmt.Sprintf("%s/reschedule/%d", request.ID, time.Now().UnixMilli())
+	msg := entityqueue.NewMessage(msgID, payload, request.Queue, nil)
+
+	q, ok := c.registry.Queue(c.topicKey)
+	if !ok {
+		return fmt.Errorf("no queue registered for topic key %s", c.topicKey)
+	}
+	topicName, ok := c.registry.TopicName(c.topicKey)
+	if !ok {
+		return fmt.Errorf("no topic name registered for topic key %s", c.topicKey)
+	}
+
+	if err := q.Publisher().PublishAfter(ctx, topicName, msg, delayMs); err != nil {
+		metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1)
+		return fmt.Errorf("ProcessController failed to reschedule process request %s: %w", request.ID, err)
+	}
+	c.logger.Infow("rescheduled latest head awaiting build slot",
+		"request_id", request.ID,
+		"queue", request.Queue,
+		"uri", request.URI,
+		"in_flight_count", inFlightCount,
+		"delay_ms", delayMs,
+	)
+	return nil
+}
+
 // loadRequest returns the request for id. A not-yet-visible row is retryable.
 func (c *Controller) loadRequest(ctx context.Context, id string) (entity.Request, error) {
 	got, err := c.store.GetRequestStore().Get(ctx, id)

stovepipe/controller/process/process_test.go

@@ -44,6 +44,14 @@ const (
 	testURI     = "git://repo/monorepo/main/abc123"
 )
 
+// rescheduledMsg matches a gate-wait re-publish: same partition, but a fresh message id —
+// re-publishing under the in-flight delivery's id would be silently deduped against its
+// still-present message-store row and lost on ack.
+func rescheduledMsg(msg entityqueue.Message) bool {
+	// Fresh non-empty id, same queue.
+	return msg.ID != testID && msg.ID != "" && msg.PartitionKey == testQueue
+}
+
 type processMocks struct {
 	reqStore      *storagemock.MockRequestStore
 	queueStore    *storagemock.MockQueueStore
@@ -74,6 +82,7 @@ func newControllerWithScope(t *testing.T, ctrl *gomock.Controller, scope tally.S
 	queue := mqmock.NewMockQueue(ctrl)
 	queue.EXPECT().Publisher().Return(m.publisher).AnyTimes()
 	registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{
+		{Key: stovepipemq.TopicKeyProcess, Name: "process", Queue: queue},
 		{Key: stovepipemq.TopicKeyBuild, Name: "build", Queue: queue},
 	})
 	require.NoError(t, err)
@@ -499,7 +508,7 @@ func TestProcess(t *testing.T) {
 			},
 		},
 		{
-			name: "latest accepted head awaits slot when gate closed",
+			name: "latest accepted head reschedules when gate closed",
 			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{
@@ -509,6 +518,52 @@ func TestProcess(t *testing.T) {
 					LastGreenURI:    "git://repo/monorepo/main/green",
 					Version:         1,
 				}, nil)
+				m.publisher.EXPECT().
+					PublishAfter(gomock.Any(), "process", gomock.Cond(rescheduledMsg), int64(5000)).
+					Return(nil)
+			},
+		},
+		{
+			name:      "gate reschedule publish error surfaces",
+			wantErr:   true,
+			wantRetry: false,
+			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,
+					InFlightCount:   1,
+					Version:         1,
+				}, nil)
+				m.publisher.EXPECT().
+					PublishAfter(gomock.Any(), "process", gomock.Cond(rescheduledMsg), int64(5000)).
+					Return(errors.New("queue down"))
+			},
+		},
+		{
+			name: "gate closed after slot claim race reschedules",
+			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(storage.ErrVersionMismatch)
+				m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{
+					Name:            testQueue,
+					LatestRequestID: testID,
+					InFlightCount:   1,
+					Version:         2,
+				}, nil)
+				m.publisher.EXPECT().
+					PublishAfter(gomock.Any(), "process", gomock.Cond(rescheduledMsg), int64(5000)).
+					Return(nil)
 			},
 		},
 		{
@@ -537,22 +592,6 @@ func TestProcess(t *testing.T) {
 				expectBuildPublish(t, m, testID)
 			},
 		},
-		{
-			name: "gate closed after reload acks without failing",
-			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(storage.ErrVersionMismatch)
-				// Reload: another admit took the last slot — gate now closed, defer (ack).
-				m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{
-					Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 2,
-				}, nil)
-			},
-		},
 		{
 			name: "reload after claim mismatch supersedes a now-stale head",
 			setup: func(m processMocks) {
@@ -765,3 +804,13 @@ func TestProcess(t *testing.T) {
 		})
 	}
 }
+
+func TestRescheduleProcessRequiresPositiveDelay(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	c, _ := newController(t, ctrl)
+
+	err := c.rescheduleProcess(context.Background(), acceptedRequest(testID), 1, 0)
+
+	require.Error(t, err)
+	assert.False(t, errs.IsRetryable(err))
+}