diff --git a/submitqueue/orchestrator/controller/dlq/batch_test.go b/submitqueue/orchestrator/controller/dlq/batch_test.go index 0c0b4c045..140f58f20 100644 --- a/submitqueue/orchestrator/controller/dlq/batch_test.go +++ b/submitqueue/orchestrator/controller/dlq/batch_test.go @@ -68,6 +68,7 @@ func TestDLQBatchController_Process_FailsAndFansOut(t *testing.T) { store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(noBatchAssociations(ctrl)).AnyTimes() c := NewDLQBatchController(zaptest.NewLogger(t).Sugar(), testScope(), staticStorageFactory{store: store}, registry, TopicKey(topickey.TopicKeyLand), "orchestrator-land-dlq") diff --git a/submitqueue/orchestrator/controller/dlq/buildsignal_test.go b/submitqueue/orchestrator/controller/dlq/buildsignal_test.go index c58b195de..de5d713dd 100644 --- a/submitqueue/orchestrator/controller/dlq/buildsignal_test.go +++ b/submitqueue/orchestrator/controller/dlq/buildsignal_test.go @@ -76,6 +76,7 @@ func TestDLQBuildSignalController_Process_FansOutToBatch(t *testing.T) { store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(noBatchAssociations(ctrl)).AnyTimes() c := NewDLQBuildSignalController(zaptest.NewLogger(t).Sugar(), testScope(), staticStorageFactory{store: store}, registry, TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq") diff --git a/submitqueue/orchestrator/controller/dlq/dlq.go b/submitqueue/orchestrator/controller/dlq/dlq.go index 31c0f7825..a7d4a98f4 100644 --- a/submitqueue/orchestrator/controller/dlq/dlq.go +++ b/submitqueue/orchestrator/controller/dlq/dlq.go @@ -235,6 +235,20 @@ func failBatch(ctx context.Context, store orchstorage.Storage, registry consumer } for _, requestID := range batch.Contains { + // A lost ack on the batch topic leaves two batch attempts for one + // request, and the other may still land it. + owner, err := owningBatch(ctx, store, requestID, batchID) + if err != nil { + return transitioned, fmt.Errorf("fan-out for batch %s: %w", batchID, err) + } + if owner != "" { + logger.Infow("dlq reconcile: request is carried by another live batch, leaving the outcome to conclude", + "request_id", requestID, + "batch_id", batchID, + "owning_batch_id", owner, + ) + continue + } if err := failRequest(ctx, store, registry, logger, requestID, lastError, metadata); err != nil { return transitioned, fmt.Errorf("fan-out for batch %s: %w", batchID, err) } diff --git a/submitqueue/orchestrator/controller/dlq/dlq_test.go b/submitqueue/orchestrator/controller/dlq/dlq_test.go index 02fd82063..85309e73b 100644 --- a/submitqueue/orchestrator/controller/dlq/dlq_test.go +++ b/submitqueue/orchestrator/controller/dlq/dlq_test.go @@ -251,6 +251,7 @@ func TestFailBatch_TransitionsAndFansOut(t *testing.T) { store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(noBatchAssociations(ctrl)).AnyTimes() _, err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "1", "", nil) require.NoError(t, err) @@ -281,6 +282,7 @@ func TestFailBatch_FailedFansOutForRepair(t *testing.T) { store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(noBatchAssociations(ctrl)).AnyTimes() _, err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "1", "", nil) require.NoError(t, err) @@ -337,11 +339,110 @@ func TestFailBatch_CancellingTransitionsToFailed(t *testing.T) { store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(noBatchAssociations(ctrl)).AnyTimes() _, err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "1", "", nil) require.NoError(t, err) } +// A lost ack leaves two batch attempts for one request. Failing one must not +// terminate a request the other, still live, batch carries. +func TestFailBatch_SkipsRequestCarriedByAnotherLiveBatch(t *testing.T) { + for _, state := range []entity.BatchState{ + entity.BatchStateCreated, + entity.BatchStateSpeculating, + entity.BatchStateLanding, + entity.BatchStateCancelling, + } { + t.Run(string(state), func(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := entity.Batch{ + ID: "1", Queue: "q", Contains: []string{"1"}, + State: entity.BatchStateSpeculating, Version: 2, + } + other := entity.Batch{ + ID: "2", Queue: "q", Contains: []string{"1"}, + State: state, Version: 1, + } + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), "1").Return(batch, nil).AnyTimes() + batchStore.EXPECT().Get(gomock.Any(), "2").Return(other, nil).AnyTimes() + batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateFailed), int32(2), int32(3)).Return(nil) + + associations := storagemock.NewMockRequestBatchStore(ctrl) + associations.EXPECT().GetByRequestID(gomock.Any(), "1").Return([]entity.RequestBatch{ + {RequestID: "1", BatchID: "1", Version: 1}, + {RequestID: "1", BatchID: "2", Version: 1}, + }, nil) + + // Request store with no EXPECTs: the request must not be touched. + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(associations).AnyTimes() + store.EXPECT().GetRequestStore().Return(storagemock.NewMockRequestStore(ctrl)).AnyTimes() + + transitioned, err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "1", "", nil) + require.NoError(t, err) + assert.True(t, transitioned) + }) + } +} + +// The batch being reconciled never counts as its own members' owner, and a +// batch that already concluded owns nothing. +func TestFailBatch_FansOutWhenNoOtherLiveBatchCarriesRequest(t *testing.T) { + tests := []struct { + name string + otherState entity.BatchState + }{ + {name: "no other batch"}, + {name: "other batch failed", otherState: entity.BatchStateFailed}, + {name: "other batch succeeded", otherState: entity.BatchStateSucceeded}, + {name: "other batch cancelled", otherState: entity.BatchStateCancelled}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + + batch := entity.Batch{ + ID: "1", Queue: "q", Contains: []string{"1"}, + State: entity.BatchStateSpeculating, Version: 2, + } + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), "1").Return(batch, nil).AnyTimes() + batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateFailed), int32(2), int32(3)).Return(nil) + + associations := []entity.RequestBatch{{RequestID: "1", BatchID: "1", Version: 1}} + if tt.otherState != "" { + other := entity.Batch{ID: "2", Queue: "q", Contains: []string{"1"}, State: tt.otherState, Version: 1} + batchStore.EXPECT().Get(gomock.Any(), "2").Return(other, nil).AnyTimes() + associations = append(associations, entity.RequestBatch{RequestID: "1", BatchID: "2", Version: 1}) + } + associationStore := storagemock.NewMockRequestBatchStore(ctrl) + associationStore.EXPECT().GetByRequestID(gomock.Any(), "1").Return(associations, nil) + + request := entity.Request{ID: "1", Queue: "q", Version: 1, State: entity.RequestStateBatched} + requestStore := storagemock.NewMockRequestStore(ctrl) + requestStore.EXPECT().Get(gomock.Any(), "1").Return(request, nil) + requestStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).Return(nil) + + registry := newTestLogRegistry(t, ctrl, 1, func(entity.RequestLog) error { return nil }) + + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(associationStore).AnyTimes() + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + + _, err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "1", "", nil) + require.NoError(t, err) + }) + } +} + func TestFailBatch_NotFoundIsNoOp(t *testing.T) { ctrl := gomock.NewController(t) diff --git a/submitqueue/orchestrator/controller/dlq/landsignal_test.go b/submitqueue/orchestrator/controller/dlq/landsignal_test.go index 938e3da8c..811249e9a 100644 --- a/submitqueue/orchestrator/controller/dlq/landsignal_test.go +++ b/submitqueue/orchestrator/controller/dlq/landsignal_test.go @@ -70,6 +70,7 @@ func TestDLQLandSignalController_Process_ReconcilesBatch(t *testing.T) { store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(noBatchAssociations(ctrl)).AnyTimes() c := NewDLQLandSignalController(zaptest.NewLogger(t).Sugar(), testScope(), staticStorageFactory{store: store}, registry, TopicKey(runwaymq.TopicKeyMergeSignal), "orchestrator-landsignal-dlq") diff --git a/submitqueue/orchestrator/controller/dlq/request.go b/submitqueue/orchestrator/controller/dlq/request.go index 6e6fc51f1..a7fa01782 100644 --- a/submitqueue/orchestrator/controller/dlq/request.go +++ b/submitqueue/orchestrator/controller/dlq/request.go @@ -157,7 +157,7 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv // stage mints one batch per delivery — and TerminateRequest does not guard // against Batched, so failing the request here would overwrite a live batch's // claim and leave it building for a request reported as failed. - owner, err := owningBatch(ctx, store, rid.ID) + owner, err := owningBatch(ctx, store, rid.ID, "") if err != nil { metrics.NamedCounter(c.metricsScope, opName, "batch_lookup_errors", 1, metrics.TagsFromContext(ctx)...) return err @@ -195,7 +195,10 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv // mid-promotion and owns it, and any other request state means the claim never // landed. A request that has since disappeared is left to failRequest, which // reports the missing row. -func owningBatch(ctx context.Context, store orchstorage.Storage, requestID string) (string, error) { +// +// excludeBatchID, when set, names a batch to ignore, so a caller reconciling +// that batch is not answered with the batch itself. +func owningBatch(ctx context.Context, store orchstorage.Storage, requestID, excludeBatchID string) (string, error) { batches, _, err := corebatch.FindByRequestID(ctx, store, requestID) if err != nil { return "", err @@ -204,6 +207,7 @@ func owningBatch(ctx context.Context, store orchstorage.Storage, requestID strin creating := "" for _, batch := range batches { switch { + case excludeBatchID != "" && batch.ID == excludeBatchID: case batch.State.IsTerminal(): case batch.State != entity.BatchStateCreating: return batch.ID, nil diff --git a/submitqueue/orchestrator/controller/dlq/speculate_test.go b/submitqueue/orchestrator/controller/dlq/speculate_test.go index a9e4479cf..f84e296f9 100644 --- a/submitqueue/orchestrator/controller/dlq/speculate_test.go +++ b/submitqueue/orchestrator/controller/dlq/speculate_test.go @@ -143,6 +143,7 @@ func TestDLQSpeculateController_Process_Attribution(t *testing.T) { store := orchstoragemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + store.EXPECT().GetRequestBatchStore().Return(noBatchAssociations(ctrl)).AnyTimes() store.EXPECT().GetQueueBatchStateStore().Return(queueBatchState).AnyTimes() var republished []string