diff --git a/stovepipe/controller/process/BUILD.bazel b/stovepipe/controller/process/BUILD.bazel index ab92f47d..0299783f 100644 --- a/stovepipe/controller/process/BUILD.bazel +++ b/stovepipe/controller/process/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "//platform/errs:go_default_library", "//platform/metrics:go_default_library", "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/entity:go_default_library", "//stovepipe/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_zap//:go_default_library", diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index 71da8a54..0f6d2b65 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -13,10 +13,9 @@ // limitations under the License. // Package process holds the process-stage queue controller. It consumes the -// request ids ingest publishes, reloads the Request from storage, and (in a -// future change) decides the build strategy by asking SourceControl how the new -// head relates to the queue's last-green URI. For now it is a thin consumer that -// reloads and logs the request, establishing the stage and its wiring. +// request ids ingest publishes, reloads the Request from storage, coalesces +// backlog to the latest head, and (in later changes) gates concurrency, decides +// build strategy, and admits winners to build. package process import ( @@ -29,12 +28,13 @@ import ( "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/zap" ) // Controller consumes ProcessRequest messages from the process stage, reloads the -// referenced Request from storage, and logs it. Implements consumer.Controller. +// referenced Request from storage, and coalesces older heads. Implements consumer.Controller. type Controller struct { logger *zap.SugaredLogger metricsScope tally.Scope @@ -63,10 +63,8 @@ func NewController( } } -// Process reloads the request referenced by the delivery and logs it. Returns nil -// to ack (success) or an error to nack (retry). A not-yet-visible request is -// retryable: ingest persists and publishes, but a stale read may not see the row -// yet, so redelivery converges. +// Process reloads the request referenced by the delivery and coalesces older heads. +// Returns nil to ack (success) or an error to nack (retry). func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { const opName = "process" @@ -82,29 +80,112 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r return fmt.Errorf("failed to deserialize process request: %w", err) } - request, err := c.store.GetRequestStore().Get(ctx, pr.Id) + request, err := c.loadRequest(ctx, pr.Id) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - if errors.Is(err, storage.ErrNotFound) { - // Retryable: the request row may not be visible yet; redelivery converges. - return errs.NewRetryableError(fmt.Errorf("request %s not found yet: %w", pr.Id, err)) + return err + } + + switch request.State { + case entity.RequestStateSuperseded, entity.RequestStateProcessing: + return nil + case entity.RequestStateAccepted: + return c.processAccepted(ctx, request) + default: + c.logger.Infow("ignored request in unexpected state", + "request_id", request.ID, + "queue", request.Queue, + "state", string(request.State), + ) + return nil + } +} + +// processAccepted coalesces older heads against queue.latest_request_id. The latest +// head is left in accepted until admit and the concurrency gate land in later PRs. +func (c *Controller) processAccepted(ctx context.Context, request entity.Request) error { + queueRow, err := c.loadQueue(ctx, request.Queue) + if err != nil { + return err + } + + if queueRow.LatestRequestID != "" { + cmp, err := entity.CompareRequestID(request.Queue, request.ID, queueRow.LatestRequestID) + if err != nil { + return fmt.Errorf("ProcessController failed to compare request ids for queue %s: %w", request.Queue, err) + } + if cmp < 0 { + if err := c.supersedeRequest(ctx, request); err != nil { + return err + } + c.logger.Infow("superseded request for newer head", + "request_id", request.ID, + "queue", request.Queue, + "latest_request_id", queueRow.LatestRequestID, + ) + return nil } - return fmt.Errorf("failed to load request %s: %w", pr.Id, err) } - c.logger.Infow("processing request", + c.logger.Infow("latest head awaiting admit", "request_id", request.ID, "queue", request.Queue, "uri", request.URI, - "state", string(request.State), - "version", request.Version, - "attempt", delivery.Attempt(), - "partition_key", msg.PartitionKey, ) - 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) + if err == nil { + return got, nil + } + if errors.Is(err, storage.ErrNotFound) { + return entity.Request{}, errs.NewRetryableError(fmt.Errorf("request %s not found yet: %w", id, err)) + } + return entity.Request{}, fmt.Errorf("ProcessController failed to load request %s: %w", id, err) +} + +// loadQueue returns the queue row for name. A not-yet-visible row is retryable. +func (c *Controller) loadQueue(ctx context.Context, name string) (entity.Queue, error) { + got, err := c.store.GetQueueStore().Get(ctx, name) + if err == nil { + return got, nil + } + if errors.Is(err, storage.ErrNotFound) { + return entity.Queue{}, errs.NewRetryableError(fmt.Errorf("queue %s not found yet: %w", name, err)) + } + return entity.Queue{}, fmt.Errorf("ProcessController failed to load queue %s: %w", name, err) +} + +// supersedeRequest CAS-marks request accepted→superseded, retrying on version conflicts. +func (c *Controller) supersedeRequest(ctx context.Context, request entity.Request) error { + reqStore := c.store.GetRequestStore() + + for { + if request.State != entity.RequestStateAccepted { + return nil + } + + updated := request + updated.State = entity.RequestStateSuperseded + newVersion := request.Version + 1 + if err := reqStore.Update(ctx, updated, request.Version, newVersion); err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + got, getErr := reqStore.Get(ctx, request.ID) + if getErr != nil { + return fmt.Errorf("ProcessController failed to reload request %s after version mismatch: %w", request.ID, getErr) + } + request = got + continue + } + return fmt.Errorf("ProcessController failed to supersede request %s: %w", request.ID, err) + } + return nil + } +} + // Name returns the controller name for logging and metrics. func (c *Controller) Name() string { return "process" diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index e9083bbc..77caaa2b 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -34,22 +34,37 @@ import ( "go.uber.org/zap" ) -const testID = "request/monorepo/main/7" +const ( + testQueue = "monorepo/main" + testID = "request/monorepo/main/7" + testURI = "git://repo/monorepo/main/abc123" +) + +type processMocks struct { + reqStore *storagemock.MockRequestStore + queueStore *storagemock.MockQueueStore +} -func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, *storagemock.MockRequestStore) { +func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processMocks) { t.Helper() - reqStore := storagemock.NewMockRequestStore(ctrl) + + m := processMocks{ + reqStore: storagemock.NewMockRequestStore(ctrl), + queueStore: storagemock.NewMockQueueStore(ctrl), + } + store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes() + store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() + store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() + c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, stovepipemq.TopicKeyProcess, "stovepipe-process") - return c, reqStore + return c, m } -// delivery wraps raw payload bytes in a MockDelivery (which satisfies consumer.Delivery). func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) consumer.Delivery { t.Helper() d := queuemock.NewMockDelivery(ctrl) - d.EXPECT().Message().Return(entityqueue.NewMessage(testID, payload, "monorepo/main", nil)).AnyTimes() + d.EXPECT().Message().Return(entityqueue.NewMessage(testID, payload, testQueue, nil)).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() return d } @@ -61,39 +76,154 @@ func processPayload(t *testing.T, id string) []byte { return b } -func TestProcess_Success(t *testing.T) { - ctrl := gomock.NewController(t) - c, reqStore := newController(t, ctrl) - reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{ID: testID, Queue: "monorepo/main", URI: "git://x", State: entity.RequestStateAccepted, Version: 1}, nil) - - require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID)))) +func acceptedRequest(id string) entity.Request { + return entity.Request{ + ID: id, + Queue: testQueue, + URI: testURI, + State: entity.RequestStateAccepted, + Version: 1, + } } -func TestProcess_NotFoundIsRetryable(t *testing.T) { - ctrl := gomock.NewController(t) - c, reqStore := newController(t, ctrl) - reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, storage.ErrNotFound) +func TestProcess(t *testing.T) { + tests := []struct { + name string + id string + setup func(m processMocks) + wantErr bool + wantRetry bool + }{ + { + name: "superseded is no-op", + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{ + ID: testID, Queue: testQueue, State: entity.RequestStateSuperseded, Version: 2, + }, nil) + }, + }, + { + name: "processing is no-op until republish lands", + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{ + ID: testID, Queue: testQueue, State: entity.RequestStateProcessing, Version: 2, + }, nil) + }, + }, + { + name: "latest accepted head awaits admit", + 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) + }, + }, + { + name: "accepted with empty latest pointer awaits admit", + 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, + Version: 1, + }, nil) + }, + }, + { + name: "older accepted head is superseded", + id: "request/monorepo/main/3", + setup: func(m processMocks) { + olderID := "request/monorepo/main/3" + m.reqStore.EXPECT().Get(gomock.Any(), olderID).Return(acceptedRequest(olderID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + Version: 1, + }, nil) + updated := acceptedRequest(olderID) + updated.State = entity.RequestStateSuperseded + m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(nil) + }, + }, + { + name: "supersede retries on version mismatch", + id: "request/monorepo/main/3", + setup: func(m processMocks) { + olderID := "request/monorepo/main/3" + m.reqStore.EXPECT().Get(gomock.Any(), olderID).Return(acceptedRequest(olderID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + Version: 1, + }, nil) + updated := acceptedRequest(olderID) + updated.State = entity.RequestStateSuperseded + m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(storage.ErrVersionMismatch) + m.reqStore.EXPECT().Get(gomock.Any(), olderID).Return(entity.Request{ + ID: olderID, Queue: testQueue, State: entity.RequestStateSuperseded, Version: 2, + }, nil) + }, + }, + { + name: "request not found is retryable", + wantErr: true, + wantRetry: true, + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, storage.ErrNotFound) + }, + }, + { + name: "queue not found is retryable", + wantErr: true, + wantRetry: true, + 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{}, storage.ErrNotFound) + }, + }, + { + name: "request storage error is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, errors.New("db down")) + }, + }, + { + name: "malformed payload is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m processMocks) {}, + }, + } - err := c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID))) - require.Error(t, err) - assert.True(t, errs.IsRetryable(err), "a not-yet-visible request must be retryable") -} + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + if tt.setup != nil { + tt.setup(m) + } -func TestProcess_StorageErrorNotRetryable(t *testing.T) { - ctrl := gomock.NewController(t) - c, reqStore := newController(t, ctrl) - reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, errors.New("db down")) - - err := c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID))) - require.Error(t, err) - assert.False(t, errs.IsRetryable(err)) -} + id := testID + if tt.id != "" { + id = tt.id + } + payload := processPayload(t, id) + if tt.name == "malformed payload is not retryable" { + payload = []byte("not-json") + } -func TestProcess_MalformedPayload(t *testing.T) { - ctrl := gomock.NewController(t) - c, _ := newController(t, ctrl) + err := c.Process(context.Background(), delivery(t, ctrl, payload)) - err := c.Process(context.Background(), delivery(t, ctrl, []byte("not-json"))) - require.Error(t, err) - assert.False(t, errs.IsRetryable(err)) + if tt.wantErr { + require.Error(t, err) + assert.Equal(t, tt.wantRetry, errs.IsRetryable(err)) + return + } + require.NoError(t, err) + }) + } }