fix(dlq): skip requests owned by another live batch when failing a batch

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")
 

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")
 

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)
 		}

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)
 

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")
 

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

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