[DRAFT/WIP] feat(stovepipe): reschedule process when concurrency gate is closed

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

@@ -24,6 +24,7 @@ import (
 	"fmt"
 
 	"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 +43,7 @@ type Controller struct {
 	metricsScope  tally.Scope
 	store         storage.Storage
 	queueConfigs  queueconfig.Store
+	registry      consumer.TopicRegistry
 	topicKey      consumer.TopicKey
 	consumerGroup string
 }
@@ -55,6 +57,7 @@ func NewController(
 	scope tally.Scope,
 	store storage.Storage,
 	queueConfigs queueconfig.Store,
+	registry consumer.TopicRegistry,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
 ) *Controller {
@@ -63,6 +66,7 @@ func NewController(
 		metricsScope:  scope.SubScope("process_controller"),
 		store:         store,
 		queueConfigs:  queueConfigs,
+		registry:      registry,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
 	}
@@ -145,31 +149,39 @@ func (c *Controller) processAccepted(ctx context.Context, request entity.Request
 	}
 
 	if queueRow.InFlightCount >= cfg.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
+		return c.rescheduleProcess(ctx, request.ID, request.Queue, cfg.GateWaitDelayMs)
 	}
 
-	return c.admitRequestToBuild(ctx, request, queueRow, cfg.MaxConcurrent)
+	return c.admitRequestToBuild(ctx, request, queueRow, cfg)
+}
+
+// rescheduleProcess acks the current delivery (by returning nil after success) and
+// re-enqueues the same ProcessRequest after a short delay so the gate can be
+// re-checked without burning MaxAttempts.
+func (c *Controller) rescheduleProcess(ctx context.Context, id, queue string, delayMs int64) error {
+	if err := c.publishProcess(ctx, id, queue, delayMs); err != nil {
+		return fmt.Errorf("ProcessController failed to reschedule process request %s: %w", id, err)
+	}
+	c.logger.Infow("rescheduled latest head awaiting build slot",
+		"request_id", id,
+		"queue", queue,
+		"delay_ms", delayMs,
+	)
+	return nil
 }
 
 // admitRequestToBuild runs the admit workflow: claim a build slot on the queue row,
 // mark the request processing with build strategy, and publish the request to build.
-func (c *Controller) admitRequestToBuild(ctx context.Context, request entity.Request, queueRow entity.Queue, maxConcurrent int32) error {
+func (c *Controller) admitRequestToBuild(ctx context.Context, request entity.Request, queueRow entity.Queue, cfg entity.QueueConfig) error {
 	for {
 		err := c.claimBuildSlot(ctx, &queueRow)
 		if err == nil {
 			break
 		}
 		if errors.Is(err, storage.ErrVersionMismatch) {
 			// claimBuildSlot reloaded queueRow; another admit may have taken the last slot.
-			if queueRow.InFlightCount >= maxConcurrent {
-				return fmt.Errorf("ProcessController gate closed for queue %s", queueRow.Name)
+			if queueRow.InFlightCount >= cfg.MaxConcurrent {
+				return c.rescheduleProcess(ctx, request.ID, request.Queue, cfg.GateWaitDelayMs)
 			}
 			continue
 		}
@@ -276,6 +288,32 @@ func (c *Controller) supersedeRequest(ctx context.Context, request entity.Reques
 	}
 }
 
+// publishProcess publishes the request ID to the process stage, partitioned by queue.
+// delayMs > 0 uses PublishAfter for gate-wait reschedules.
+func (c *Controller) publishProcess(ctx context.Context, id, queue string, delayMs int64) error {
+	payload, err := stovepipemq.Marshal(&stovepipemq.ProcessRequest{Id: id})
+	if err != nil {
+		return fmt.Errorf("failed to serialize process request: %w", err)
+	}
+
+	msg := entityqueue.NewMessage(id, payload, 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)
+	}
+
+	publisher := q.Publisher()
+	if delayMs > 0 {
+		return publisher.PublishAfter(ctx, topicName, msg, delayMs)
+	}
+	return publisher.Publish(ctx, topicName, msg)
+}
+
 // 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,7 @@ const (
 type processMocks struct {
 	reqStore   *storagemock.MockRequestStore
 	queueStore *storagemock.MockQueueStore
+	publisher  *mqmock.MockPublisher
 }
 
 func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processMocks) {
@@ -52,13 +53,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
 }
 
@@ -149,7 +159,24 @@ 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{
+					Name:            testQueue,
+					LatestRequestID: testID,
+					InFlightCount:   1,
+					Version:         1,
+				}, nil)
+				m.publisher.EXPECT().
+					PublishAfter(gomock.Any(), "process", gomock.Any(), 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{
@@ -158,6 +185,35 @@ func TestProcess(t *testing.T) {
 					InFlightCount:   1,
 					Version:         1,
 				}, nil)
+				m.publisher.EXPECT().
+					PublishAfter(gomock.Any(), "process", gomock.Any(), 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.Any(), int64(5000)).
+					Return(nil)
 			},
 		},
 		{