diff --git a/submitqueue/entity/batch.go b/submitqueue/entity/batch.go index 39e21e5d..45164ead 100644 --- a/submitqueue/entity/batch.go +++ b/submitqueue/entity/batch.go @@ -62,6 +62,17 @@ func (s BatchState) IsTerminal() bool { } } +var cancellableBatchStates = map[BatchState]bool{ + BatchStateCreated: true, + BatchStateSpeculating: true, + BatchStateCancelling: true, +} + +// IsCancellable returns true if cancellation should transition or republish a batch in this state. +func (s BatchState) IsCancellable() bool { + return cancellableBatchStates[s] +} + // IsBatchStateHalted returns true if the batch is either terminal or in the process of being cancelled. // Forward-progress controllers (build, buildsignal, speculate, merge) use this to short-circuit // work for batches that the user has asked to cancel — even though Cancelling is non-terminal, no diff --git a/submitqueue/entity/batch_test.go b/submitqueue/entity/batch_test.go index ebdfcd04..ea6a2f8f 100644 --- a/submitqueue/entity/batch_test.go +++ b/submitqueue/entity/batch_test.go @@ -45,6 +45,18 @@ func TestBatchState_IsTerminal(t *testing.T) { } } +func TestIsCancellable(t *testing.T) { + assert.True(t, BatchStateCreated.IsCancellable()) + assert.True(t, BatchStateSpeculating.IsCancellable()) + assert.True(t, BatchStateCancelling.IsCancellable()) + assert.False(t, BatchStateUnknown.IsCancellable()) + assert.False(t, BatchStateCreating.IsCancellable()) + assert.False(t, BatchStateMerging.IsCancellable()) + assert.False(t, BatchStateSucceeded.IsCancellable()) + assert.False(t, BatchStateFailed.IsCancellable()) + assert.False(t, BatchStateCancelled.IsCancellable()) +} + func TestActiveBatchStates_ExcludesCreating(t *testing.T) { assert.NotContains(t, ActiveBatchStates(), BatchStateCreating) } diff --git a/submitqueue/orchestrator/controller/cancel/cancel.go b/submitqueue/orchestrator/controller/cancel/cancel.go index aed3cb62..54a411de 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel.go +++ b/submitqueue/orchestrator/controller/cancel/cancel.go @@ -23,13 +23,10 @@ // RequestStatusCancelled log entry. This path is fully owned by the cancel // controller. // -// - The request is already part of an active batch — the controller performs -// a single intent CAS on the batch (advancing it to BatchStateCancelling) -// and hands off to the speculate controller by publishing the batch ID to -// TopicKeySpeculate. The speculate controller then owns: cancelling any -// in-flight Build entity for the batch, fanning out to dependents, the -// terminal CAS to BatchStateCancelled, and publishing to conclude. Cancel -// does no terminal write and no downstream fan-out on the batch path. +// - The request is part of one or more published batch attempts — the controller +// records cancellation intent on every Created or Speculating attempt and +// hands each one to speculate. Merging and terminal attempts retain their +// existing outcome, which conclude uses to reconcile the request. // // The split exists so that the terminal write and the work that must precede // it (cancelling builds, respeculating dependents) live in the same controller @@ -40,11 +37,9 @@ // its terminal state. // // The controller is idempotent: re-delivery of the same CancelRequest after -// the terminal request transition is a no-op; re-delivery after the -// Cancelling write skips the mark-cancelling step and proceeds straight to -// the batch lookup. On the batch path, re-delivery against an already -// Cancelling batch re-publishes to TopicKeySpeculate (a cheap no-op nudge -// the speculate controller absorbs). +// the terminal request transition is a no-op. Re-delivery after a Cancelling +// write skips that CAS, finds every batch ID containing the request again, and +// republishes matching attempts already in BatchStateCancelling. // // Concurrent producers surface as the intrinsically retryable // storage.ErrVersionMismatch; the controller returns the wrapped error as-is @@ -57,6 +52,7 @@ package cancel import ( "context" "fmt" + "sort" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" @@ -144,16 +140,46 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return err } - // Look for an active batch that already contains this request. - batch, found, err := c.findActiveBatch(ctx, request) + // Find every eligible batch containing this request. Retries may create multiple batch IDs for the same request, and each must be handled. + batches, err := c.findBatches(ctx, request) if err != nil { return err } - if !found { + // Return the first failure so the existing linear error-classification chain remains intact. + // Matches are sorted by ID below, making the selected error deterministic while every failure is still logged and counted. + var firstErr error + foundApplicableBatch := false + for _, batch := range batches { + switch { + case batch.State.IsCancellable(): + foundApplicableBatch = true + if err := c.cancelBatch(ctx, batch); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "batch_cancel_errors", 1) + c.logger.Errorw("failed to cancel batch", + "batch_id", batch.ID, + "batch_state", string(batch.State), + "error", err, + ) + if firstErr == nil { + firstErr = err + } + } + case batch.State == entity.BatchStateMerging: + // Merge owns the outcome once it has started. Conclude will reconcile the request with that outcome. + foundApplicableBatch = true + metrics.NamedCounter(c.metricsScope, opName, "batch_merging", 1) + case batch.State.IsTerminal(): + // The terminal batch outcome wins; conclude may not have reconciled the request yet. + foundApplicableBatch = true + metrics.NamedCounter(c.metricsScope, opName, "batch_already_terminal", 1) + } + } + + if !foundApplicableBatch { return c.cancelRequest(ctx, request, cancelReq.Reason) } - return c.cancelBatch(ctx, batch) + return firstErr } // markCancelling transitions the request to RequestStateCancelling (intent) if it @@ -182,31 +208,42 @@ func (c *Controller) markCancelling(ctx context.Context, request entity.Request) return request, nil } -// findActiveBatch scans all active batches in the request's queue for one whose -// Contains list includes the request. Returns (batch, true, nil) on a hit, -// (zero, false, nil) when the request is not yet batched, and any storage -// error otherwise. -// -// BatchStateCancelling is included in the active-state list so an idempotent -// redelivery of the cancel message (the prior pass wrote the intent but the -// speculate hand-off publish failed) still resolves the batch and re-attempts -// the publish. -func (c *Controller) findActiveBatch(ctx context.Context, request entity.Request) (entity.Batch, bool, error) { - // TODO: Scans all the batches in flight - make it more efficient? - active, err := c.store.GetBatchStore().GetByQueueAndStates(ctx, request.Queue, entity.ActiveBatchStates()) +// findBatches returns every eligible batch attempt containing the request. +// The current state-based store lookup is the adaptation point for the publication-aware lookup being added separately. +func (c *Controller) findBatches(ctx context.Context, request entity.Request) ([]entity.Batch, error) { + batches, err := c.store.GetBatchStore().GetByQueueAndStates(ctx, request.Queue, cancellationBatchStates()) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "batch_store_errors", 1) - return entity.Batch{}, false, fmt.Errorf("failed to get active batches for queue=%s: %w", request.Queue, err) + return nil, fmt.Errorf("failed to get batches for queue=%s: %w", request.Queue, err) } - for _, b := range active { - for _, rid := range b.Contains { - if rid == request.ID { - return b, true, nil + var matches []entity.Batch + for _, batch := range batches { + for _, requestID := range batch.Contains { + if requestID == request.ID { + matches = append(matches, batch) + break } } } - return entity.Batch{}, false, nil + // The batches are independent, but deterministic order stabilizes logs, tests, and first-error selection. + sort.Slice(matches, func(i, j int) bool { + return matches[i].ID < matches[j].ID + }) + return matches, nil +} + +func cancellationBatchStates() []entity.BatchState { + // Creating attempts are unpublished and may not have a reverse-index row, so cancellation ignores them. + return []entity.BatchState{ + entity.BatchStateCreated, + entity.BatchStateSpeculating, + entity.BatchStateMerging, + entity.BatchStateCancelling, + entity.BatchStateSucceeded, + entity.BatchStateFailed, + entity.BatchStateCancelled, + } } // cancelRequest drives the terminal transition (Cancelling → Cancelled) for a diff --git a/submitqueue/orchestrator/controller/cancel/cancel_test.go b/submitqueue/orchestrator/controller/cancel/cancel_test.go index 79724a99..731eb066 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel_test.go +++ b/submitqueue/orchestrator/controller/cancel/cancel_test.go @@ -348,10 +348,160 @@ func TestProcess_BatchPath_HandsOffToSpeculate(t *testing.T) { assert.Equal(t, []pubRec{{topic: "speculate", msgID: "q/batch/1"}}, records) } +func TestProcess_CancelsEveryApplicableBatch(t *testing.T) { + ctrl := gomock.NewController(t) + registry, publisher := newRegistry(t, ctrl) + + request := entity.Request{ID: "q/1", Queue: "q", State: entity.RequestStateBatched, Version: 2} + batch1 := entity.Batch{ID: "q/batch/1", Queue: "q", Contains: []string{request.ID}, State: entity.BatchStateCreated, Version: 1} + batch2 := entity.Batch{ID: "q/batch/2", Queue: "q", Contains: []string{request.ID}, State: entity.BatchStateSpeculating, Version: 3} + terminalBatch := entity.Batch{ID: "q/batch/3", Queue: "q", Contains: []string{request.ID}, State: entity.BatchStateSucceeded, Version: 4} + + requestStore := storagemock.NewMockRequestStore(ctrl) + requestStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil) + requestStore.EXPECT().UpdateState(gomock.Any(), request.ID, int32(2), int32(3), entity.RequestStateCancelling).Return(nil) + + var operations []string + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().GetByQueueAndStates(gomock.Any(), request.Queue, cancellationBatchStates()).Return([]entity.Batch{ + terminalBatch, + batch2, + batch1, + }, nil) + batchStore.EXPECT().UpdateState(gomock.Any(), batch1.ID, int32(1), int32(2), entity.BatchStateCancelling).DoAndReturn( + func(context.Context, string, int32, int32, entity.BatchState) error { + operations = append(operations, "update:"+batch1.ID) + return nil + }, + ) + batchStore.EXPECT().UpdateState(gomock.Any(), batch2.ID, int32(3), int32(4), entity.BatchStateCancelling).DoAndReturn( + func(context.Context, string, int32, int32, entity.BatchState) error { + operations = append(operations, "update:"+batch2.ID) + return nil + }, + ) + publisher.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + operations = append(operations, "publish:"+msg.ID) + return nil + }, + ).Times(2) + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + + controller := newController(t, store, registry) + assert.NoError(t, controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, request.ID, ""), request.ID))) + assert.Equal(t, []string{ + "update:q/batch/1", + "publish:q/batch/1", + "update:q/batch/2", + "publish:q/batch/2", + }, operations) +} + +func TestProcess_BatchFailureDoesNotPreventLaterCancellation(t *testing.T) { + ctrl := gomock.NewController(t) + registry, publisher := newRegistry(t, ctrl) + + request := entity.Request{ID: "q/1", Queue: "q", State: entity.RequestStateBatched, Version: 2} + batch1 := entity.Batch{ID: "q/batch/1", Queue: "q", Contains: []string{request.ID}, State: entity.BatchStateCreated, Version: 1} + batch2 := entity.Batch{ID: "q/batch/2", Queue: "q", Contains: []string{request.ID}, State: entity.BatchStateCreated, Version: 2} + storeErr := fmt.Errorf("storage failed") + + requestStore := storagemock.NewMockRequestStore(ctrl) + requestStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil) + requestStore.EXPECT().UpdateState(gomock.Any(), request.ID, int32(2), int32(3), entity.RequestStateCancelling).Return(nil) + + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().GetByQueueAndStates(gomock.Any(), request.Queue, cancellationBatchStates()).Return([]entity.Batch{batch2, batch1}, nil) + batchStore.EXPECT().UpdateState(gomock.Any(), batch1.ID, int32(1), int32(2), entity.BatchStateCancelling).Return(storeErr) + batchStore.EXPECT().UpdateState(gomock.Any(), batch2.ID, int32(2), int32(3), entity.BatchStateCancelling).Return(nil) + publisher.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + assert.Equal(t, batch2.ID, msg.ID) + return nil + }, + ) + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + + controller := newController(t, store, registry) + err := controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, request.ID, ""), request.ID)) + assert.ErrorContains(t, err, "failed to mark batch q/batch/1 as cancelling") +} + +func TestProcess_NonCancellableBatchSuppressesRequestCancellation(t *testing.T) { + tests := []struct { + name string + state entity.BatchState + }{ + {name: "merging", state: entity.BatchStateMerging}, + {name: "succeeded", state: entity.BatchStateSucceeded}, + {name: "failed", state: entity.BatchStateFailed}, + {name: "cancelled", state: entity.BatchStateCancelled}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + registry, _ := newRegistry(t, ctrl) + + request := entity.Request{ID: "q/1", Queue: "q", State: entity.RequestStateBatched, Version: 2} + batch := entity.Batch{ID: "q/batch/1", Queue: "q", Contains: []string{request.ID}, State: tt.state, Version: 4} + + requestStore := storagemock.NewMockRequestStore(ctrl) + requestStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil) + requestStore.EXPECT().UpdateState(gomock.Any(), request.ID, int32(2), int32(3), entity.RequestStateCancelling).Return(nil) + + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().GetByQueueAndStates(gomock.Any(), request.Queue, cancellationBatchStates()).Return([]entity.Batch{batch}, nil) + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + + controller := newController(t, store, registry) + assert.NoError(t, controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, request.ID, ""), request.ID))) + }) + } +} + +func TestProcess_BatchedWithoutMatchCancelsRequest(t *testing.T) { + ctrl := gomock.NewController(t) + registry, publisher := newRegistry(t, ctrl) + publisher.EXPECT().Publish(gomock.Any(), "log", gomock.Any()).Return(nil) + + request := entity.Request{ID: "q/1", Queue: "q", State: entity.RequestStateBatched, Version: 2} + cancellingRequest := request + cancellingRequest.State = entity.RequestStateCancelling + cancellingRequest.Version = 3 + requestStore := storagemock.NewMockRequestStore(ctrl) + gomock.InOrder( + requestStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil), + requestStore.EXPECT().UpdateState(gomock.Any(), request.ID, int32(2), int32(3), entity.RequestStateCancelling).Return(nil), + requestStore.EXPECT().Get(gomock.Any(), request.ID).Return(cancellingRequest, nil), + requestStore.EXPECT().UpdateState(gomock.Any(), request.ID, int32(3), int32(4), entity.RequestStateCancelled).Return(nil), + ) + + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().GetByQueueAndStates(gomock.Any(), request.Queue, cancellationBatchStates()).Return(nil, nil) + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + + controller := newController(t, store, registry) + assert.NoError(t, controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, request.ID, "stop"), request.ID))) +} + // TestProcess_BatchAlreadyCancelling_RepublishesToSpeculate covers the // idempotent redelivery path on the batch flow: a prior pass wrote the // intent CAS but the speculate publish failed. The cancel controller must -// observe the active (Cancelling) batch via findActiveBatch, skip the +// observe the Cancelling batch via findBatches, skip the // intent CAS, and re-publish the batch ID to TopicKeySpeculate so the // speculate controller can pick the work up. func TestProcess_BatchAlreadyCancelling_RepublishesToSpeculate(t *testing.T) { @@ -449,3 +599,62 @@ func TestProcess_RequestStoreError(t *testing.T) { err := controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "q/1", ""), "q/1")) require.Error(t, err) } + +func TestFindBatches(t *testing.T) { + request := entity.Request{ID: "q/1", Queue: "q"} + matching1 := entity.Batch{ID: "q/batch/1", Contains: []string{request.ID}} + matching2 := entity.Batch{ID: "q/batch/2", Contains: []string{"q/other", request.ID}} + storeErr := fmt.Errorf("storage failed") + + tests := []struct { + name string + mockFunc func(*storagemock.MockBatchStore) + want []entity.Batch + errMsg string + }{ + { + name: "lookup failure", + mockFunc: func(store *storagemock.MockBatchStore) { + store.EXPECT().GetByQueueAndStates(gomock.Any(), request.Queue, cancellationBatchStates()).Return(nil, storeErr) + }, + errMsg: "failed to get batches", + }, + { + name: "nonmatching batches", + mockFunc: func(store *storagemock.MockBatchStore) { + store.EXPECT().GetByQueueAndStates(gomock.Any(), request.Queue, cancellationBatchStates()).Return( + []entity.Batch{{ID: "q/batch/other", Contains: []string{"q/other"}}}, nil, + ) + }, + }, + { + name: "matching batches are sorted", + mockFunc: func(store *storagemock.MockBatchStore) { + store.EXPECT().GetByQueueAndStates(gomock.Any(), request.Queue, cancellationBatchStates()).Return( + []entity.Batch{matching2, {ID: "q/batch/other", Contains: []string{"q/other"}}, matching1}, nil, + ) + }, + want: []entity.Batch{matching1, matching2}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + batchStore := storagemock.NewMockBatchStore(ctrl) + tt.mockFunc(batchStore) + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore) + + controller := &Controller{metricsScope: tally.NoopScope, store: store} + got, err := controller.findBatches(context.Background(), request) + if tt.errMsg != "" { + assert.ErrorContains(t, err, tt.errMsg) + return + } + assert.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +}