From 9bc9a5f54e3d8febfa8850fc7eb94011973ebfee Mon Sep 17 00:00:00 2001 From: "kevin.new" Date: Wed, 15 Jul 2026 22:02:55 +0000 Subject: [PATCH] fix(orchestrator): add Validated state transition and log in mergeconflictsignal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../mergeconflictsignal.go | 19 ++++++++++-- .../mergeconflictsignal_test.go | 30 ++++++++++++++----- 2 files changed, 38 insertions(+), 11 deletions(-) diff --git a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go index d04aaa26..66bbc9bd 100644 --- a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go +++ b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go @@ -121,8 +121,6 @@ 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) @@ -130,12 +128,27 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r 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, ) diff --git a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal_test.go b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal_test.go index fe9c48ad..56f289d3 100644 --- a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal_test.go +++ b/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) }