fix(orchestrator): add Validated state transition and log in mergeconflictsignal

The mergeconflictsignal controller was publishing directly to the batch
topic on success without transitioning the request to Validated or
recording a log entry. This adds the missing CAS Started→Validated
state transition and publishes a RequestStatusValidated log entry
before forwarding to batch.

Author: kevinlnew

submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go

@@ -121,21 +121,34 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 			"request_id", request.ID,
 			"reason", result.Reason,
 		)
-		// Expected terminal outcome, not a failure: mark the request Error inline
-		// and ack.
 		if err := c.failRequest(ctx, request, result.Reason); err != nil {
 			metrics.NamedCounter(c.metricsScope, opName, "fail_errors", 1)
 			return fmt.Errorf("failed to fail request %s: %w", request.ID, err)
 		}
 		return nil
 	}
 
+	// Advance the request to Validated now that the merge-conflict check passed.
+	newVersion := request.Version + 1
+	if err := c.store.GetRequestStore().UpdateState(ctx, request.ID, request.Version, newVersion, entity.RequestStateValidated); err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "state_errors", 1)
+		return fmt.Errorf("failed to update request %s state to validated: %w", request.ID, err)
+	}
+	request.Version = newVersion
+	request.State = entity.RequestStateValidated
+
+	logEntry := entity.NewRequestLog(request.ID, entity.RequestStatusValidated, request.Version, "", nil)
+	if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID); err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "log_errors", 1)
+		return fmt.Errorf("failed to publish request log for %s: %w", request.ID, err)
+	}
+
 	if err := c.publishRequestID(ctx, topickey.TopicKeyBatch, request.ID, request.Queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish to batch: %w", err)
 	}
 
-	c.logger.Infow("published request to batch",
+	c.logger.Infow("request validated and published to batch",
 		"request_id", request.ID,
 		"topic_key", topickey.TopicKeyBatch,
 	)

submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal_test.go

@@ -57,24 +57,28 @@ func TestProcess_MergeablePublishesToBatch(t *testing.T) {
 	reqStore := storagemock.NewMockRequestStore(ctrl)
 	reqStore.EXPECT().Get(gomock.Any(), testRequestID).Return(
 		entity.Request{ID: testRequestID, Queue: testQueue, State: entity.RequestStateStarted, Version: 1}, nil)
+	reqStore.EXPECT().UpdateState(gomock.Any(), testRequestID, int32(1), int32(2), entity.RequestStateValidated).Return(nil)
 
 	store := storagemock.NewMockStorage(ctrl)
 	store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes()
 
-	var gotTopic string
-	var gotPayload []byte
+	var gotTopics []string
+	var gotPayloads [][]byte
 	pub := queuemock.NewMockPublisher(ctrl)
 	pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
 		func(_ context.Context, topic string, msg entityqueue.Message) error {
-			gotTopic = topic
-			gotPayload = msg.Payload
+			gotTopics = append(gotTopics, topic)
+			gotPayloads = append(gotPayloads, msg.Payload)
 			return nil
 		},
-	)
+	).Times(2)
 	q := queuemock.NewMockQueue(ctrl)
 	q.EXPECT().Publisher().Return(pub).AnyTimes()
 	registry, err := consumer.NewTopicRegistry(
-		[]consumer.TopicConfig{{Key: topickey.TopicKeyBatch, Name: "batch", Queue: q}},
+		[]consumer.TopicConfig{
+			{Key: topickey.TopicKeyBatch, Name: "batch", Queue: q},
+			{Key: topickey.TopicKeyLog, Name: "log", Queue: q},
+		},
 	)
 	require.NoError(t, err)
 
@@ -85,8 +89,18 @@ func TestProcess_MergeablePublishesToBatch(t *testing.T) {
 	msg := entityqueue.NewMessage(testRequestID, resultPayload(t, res), testQueue, nil)
 	require.NoError(t, controller.Process(context.Background(), newDelivery(ctrl, msg)))
 
-	assert.Equal(t, "batch", gotTopic)
-	rid, err := entity.RequestIDFromBytes(gotPayload)
+	require.Len(t, gotTopics, 2)
+
+	// First publish: validated log entry.
+	assert.Equal(t, "log", gotTopics[0])
+	logEntry, err := entity.RequestLogFromBytes(gotPayloads[0])
+	require.NoError(t, err)
+	assert.Equal(t, entity.RequestStatusValidated, logEntry.Status)
+	assert.Equal(t, int32(2), logEntry.RequestVersion)
+
+	// Second publish: request ID to batch topic.
+	assert.Equal(t, "batch", gotTopics[1])
+	rid, err := entity.RequestIDFromBytes(gotPayloads[1])
 	require.NoError(t, err)
 	assert.Equal(t, testRequestID, rid.ID)
 }