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

service/stovepipe/server/main.go

@@ -210,7 +210,7 @@ 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(), registry, stovepipemq.TopicKeyProcess, "stovepipe-process")
 	if err := primaryConsumer.Register(processController); err != nil {
 		return fmt.Errorf("failed to register process controller: %w", err)
 	}

stovepipe/controller/process/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     importpath = "github.com/uber/submitqueue/stovepipe/controller/process",
     visibility = ["//visibility:public"],
     deps = [
+        "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/errs:go_default_library",
         "//platform/metrics:go_default_library",

stovepipe/controller/process/process.go

@@ -22,8 +22,10 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/uber-go/tally"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/errs"
 	"github.com/uber/submitqueue/platform/metrics"
@@ -42,6 +44,7 @@ type Controller struct {
 	metricsScope  tally.Scope
 	store         storage.Storage
 	queueConfigs  queueconfig.Store
+	registry      consumer.TopicRegistry
 	topicKey      consumer.TopicKey
 	consumerGroup string
 }
@@ -58,6 +61,7 @@ func NewController(
 	scope tally.Scope,
 	store storage.Storage,
 	queueConfigs queueconfig.Store,
+	registry consumer.TopicRegistry,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
 ) *Controller {
@@ -66,6 +70,7 @@ func NewController(
 		metricsScope:  scope.SubScope("process_controller"),
 		store:         store,
 		queueConfigs:  queueConfigs,
+		registry:      registry,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
 	}
@@ -142,7 +147,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
@@ -172,18 +177,11 @@ 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 {
 	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)
 		}
 
 		err := c.claimBuildSlot(ctx, &queueRow)
@@ -349,6 +347,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

@@ -42,9 +42,18 @@ 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
+	publisher  *mqmock.MockPublisher
 }
 
 func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processMocks) {
@@ -53,13 +62,22 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processM
 	m := processMocks{
 		reqStore:   storagemock.NewMockRequestStore(ctrl),
 		queueStore: storagemock.NewMockQueueStore(ctrl),
+		publisher:  mqmock.NewMockPublisher(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")
+	queue := mqmock.NewMockQueue(ctrl)
+	queue.EXPECT().Publisher().Return(m.publisher).AnyTimes()
+
+	registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{
+		{Key: stovepipemq.TopicKeyProcess, Name: "process", Queue: queue},
+	})
+	require.NoError(t, err)
+
+	c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, queueconfigdefault.NewStore(), registry, stovepipemq.TopicKeyProcess, "stovepipe-process")
 	return c, m
 }
 
@@ -159,7 +177,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{
@@ -168,6 +186,52 @@ func TestProcess(t *testing.T) {
 					InFlightCount:   1,
 					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)
 			},
 		},
 		{
@@ -195,22 +259,6 @@ func TestProcess(t *testing.T) {
 				m.reqStore.EXPECT().Update(gomock.Any(), updatedReq, int32(1), int32(2)).Return(nil)
 			},
 		},
-		{
-			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) {
@@ -390,3 +438,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))
+}