From 5b651c5d29d66410ff024f1d716d40ef766475ec Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Mon, 13 Jul 2026 15:59:24 +0000 Subject: [PATCH] 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. --- service/stovepipe/server/main.go | 2 +- stovepipe/controller/process/BUILD.bazel | 1 + stovepipe/controller/process/process.go | 63 ++++++++++--- stovepipe/controller/process/process_test.go | 94 ++++++++++++++++---- 4 files changed, 129 insertions(+), 31 deletions(-) diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 22c53a60..236191e7 100644 --- a/service/stovepipe/server/main.go +++ b/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) } diff --git a/stovepipe/controller/process/BUILD.bazel b/stovepipe/controller/process/BUILD.bazel index a856f647..a99ff0b5 100644 --- a/stovepipe/controller/process/BUILD.bazel +++ b/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", diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index dd80fdd1..9e9946b5 100644 --- a/stovepipe/controller/process/process.go +++ b/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) diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index 610223a0..3cd5b363 100644 --- a/stovepipe/controller/process/process_test.go +++ b/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)) +}