diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index 6317bc646..1ec9c7da1 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -14,8 +14,7 @@ // Package process holds the process-stage queue controller. It consumes request // ids from ingest, reloads the Request from storage, coalesces older heads, and -// admits the latest head when a build slot is open. Build queue publish lands in -// a follow-up PR. +// admits the latest head when a build slot is open and dispatches its build. package process import ( @@ -124,22 +123,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er switch request.State { case entity.RequestStateProcessing: - if err := c.persistStateLog(ctx, store, request, entity.RequestOutcomeReasonUnknown); err != nil { - return err - } - // Announce here as well as at admit: this is the only path a redelivery - // takes once the transition is durable, so an admit that failed after - // persisting would otherwise lose the start event for good. The event id - // is derived from the transition, so a repeat carries the id the first - // attempt would have and a consumer deduplicates on it. - if err := c.publishHookEvent(ctx, request, hookevent.NewValidationRepositoryStarted(request)); err != nil { - return err - } - if err := c.publishBuild(ctx, request.ID, request.Queue); err != nil { - metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1, metrics.TagsFromContext(ctx)...) - return fmt.Errorf("failed to publish request %s to build: %w", request.ID, err) - } - return nil + return c.handoffBuild(ctx, store, request) case entity.RequestStateSuperseded: return c.persistStateLog(ctx, store, request, entity.RequestOutcomeReasonSupersededByNewerHead) case entity.RequestStateSucceeded, entity.RequestStateFailed, entity.RequestStateCancelled: @@ -191,43 +175,6 @@ func (c *Controller) processAccepted(ctx context.Context, store storage.Storage, return fmt.Errorf("failed to load queue config for %s: %w", request.Queue, err) } - return c.admitLatestHead(ctx, store, delivery, request, queueRow, cfg) -} - -// coalesce supersedes request when a newer head exists (RFC process step 5), returning -// true so the caller acks. It returns false when request is still the latest head and -// should proceed to the gate. Superseding consumes no build slot. -func (c *Controller) coalesce(ctx context.Context, store storage.Storage, request entity.Request, latestRequestID string) (bool, error) { - cmp, err := entity.CompareRequestID(request.ID, latestRequestID) - if err != nil { - return false, fmt.Errorf("failed to compare request ids for queue %s: %w", request.Queue, err) - } - if cmp >= 0 { - return false, nil - } - request, err = c.supersedeRequest(ctx, store, request) - if err != nil { - metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) - return false, err - } - if request.State != entity.RequestStateSuperseded { - return true, nil - } - if err := c.persistStateLog(ctx, store, request, entity.RequestOutcomeReasonSupersededByNewerHead); err != nil { - return false, err - } - metrics.NamedCounter(c.metricsScope, _opName, "superseded", 1, metrics.TagsFromContext(ctx)...) - c.logger.Infow("superseded request for newer head", - "request_id", request.ID, - "queue", request.Queue, - "latest_request_id", latestRequestID, - ) - return true, nil -} - -// admitLatestHead holds the delivery while the build-slot gate is closed; otherwise it marks the -// request processing, counts its build slot, and publishes it to build. -func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage, delivery consumer.Delivery, request entity.Request, queueRow entity.Queue, cfg entity.QueueConfig) error { if queueRow.InFlightCount >= cfg.MaxConcurrent { return c.holdForBuildSlot(ctx, delivery, request, queueRow.InFlightCount, cfg.GateWaitDelayMs) } @@ -250,19 +197,10 @@ func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage, return err } - if err := c.persistStateLog(ctx, store, request, entity.RequestOutcomeReasonUnknown); err != nil { + if err := c.handoffBuild(ctx, store, request); err != nil { return err } - if err := c.publishHookEvent(ctx, request, hookevent.NewValidationRepositoryStarted(request)); err != nil { - return err - } - - if err := c.publishBuild(ctx, request.ID, request.Queue); err != nil { - metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1, metrics.TagsFromContext(ctx)...) - return fmt.Errorf("failed to publish request %s to build: %w", request.ID, err) - } - metrics.NamedCounter(c.metricsScope, _opName, "admitted", 1, metrics.TagsFromContext(ctx, metrics.NewTag("strategy", string(request.BuildStrategy)))..., ) @@ -276,6 +214,52 @@ func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage, return nil } +// handoffBuild repeats the durable start announcement and build dispatch on redelivery. +func (c *Controller) handoffBuild(ctx context.Context, store storage.Storage, request entity.Request) error { + if err := c.persistStateLog(ctx, store, request, entity.RequestOutcomeReasonUnknown); err != nil { + return err + } + if err := c.publishHookEvent(ctx, request, hookevent.NewValidationRepositoryStarted(request)); err != nil { + return err + } + if err := c.publishBuild(ctx, request.ID, request.Queue); err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1, metrics.TagsFromContext(ctx)...) + return fmt.Errorf("failed to publish request %s to build: %w", request.ID, err) + } + return nil +} + +// coalesce supersedes request when a newer head exists (RFC process step 5), returning +// true so the caller acks. It returns false when request is still the latest head and +// should proceed to the gate. Superseding consumes no build slot. +func (c *Controller) coalesce(ctx context.Context, store storage.Storage, request entity.Request, latestRequestID string) (bool, error) { + cmp, err := entity.CompareRequestID(request.ID, latestRequestID) + if err != nil { + return false, fmt.Errorf("failed to compare request ids for queue %s: %w", request.Queue, err) + } + if cmp >= 0 { + return false, nil + } + request, err = c.supersedeRequest(ctx, store, request) + if err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) + return false, err + } + if request.State != entity.RequestStateSuperseded { + return true, nil + } + if err := c.persistStateLog(ctx, store, request, entity.RequestOutcomeReasonSupersededByNewerHead); err != nil { + return false, err + } + metrics.NamedCounter(c.metricsScope, _opName, "superseded", 1, metrics.TagsFromContext(ctx)...) + c.logger.Infow("superseded request for newer head", + "request_id", request.ID, + "queue", request.Queue, + "latest_request_id", latestRequestID, + ) + return true, nil +} + func (c *Controller) persistStateLog(ctx context.Context, store storage.Storage, request entity.Request, reason entity.RequestOutcomeReason) error { log := requestlog.NewRequestStateLog(request, reason) if err := c.materializer.PersistLog(ctx, store, log); err != nil { diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index 2f5e87f9a..c4d64e753 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -158,17 +158,25 @@ func acceptedRequest(id string) entity.Request { func TestProcessBuildPublishRequiresRegisteredTopic(t *testing.T) { ctrl := gomock.NewController(t) c, m := newController(t, ctrl) - c.registry = consumer.TopicRegistry{} + queue := mqmock.NewMockQueue(ctrl) + queue.EXPECT().Publisher().Return(m.publisher) + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: basehook.TopicKeyHook, Name: "stovepipe-hook", Queue: queue}, + }) + require.NoError(t, err) + c.registry = registry request := entity.Request{ ID: testID, Queue: testQueue, State: entity.RequestStateProcessing, Version: 2, } m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) - expectProcessingLog(t, m, request) + logCall := expectProcessingLog(t, m, request) + expectStartValidationAnnounce(t, m, request.ID).After(logCall) - err := c.Process(queueContext(testQueue), delivery(t, ctrl, processPayload(t, testID))) + err = c.Process(queueContext(testQueue), delivery(t, ctrl, processPayload(t, testID))) require.Error(t, err) + assert.Contains(t, err.Error(), "no queue registered for topic key build") assert.False(t, errs.IsRetryable(err)) }