feat(orchestrator): emit batched request log from batch controller (#287)

## Summary
Previously the batch controller was not publishing a request log on
completion. This is inconsistent with the other controllers, which
publish a terminal request log when its action is complete.

This PR emits a request log after the batch is persisted and before
publishing to score.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

## Test Plan
✅ make fmt && make build && make test && make check-mocks && make
e2e-test

## Issues


## Stack
1. @ #287
1. #288
1. #289

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: albertywu

submitqueue/orchestrator/controller/batch/BUILD.bazel

@@ -10,6 +10,7 @@ go_library(
         "//platform/consumer",
         "//platform/extension/counter",
         "//platform/metrics",
+        "//submitqueue/core/request",
         "//submitqueue/core/topickey",
         "//submitqueue/entity",
         "//submitqueue/extension/conflict",

submitqueue/orchestrator/controller/batch/batch.go

@@ -24,6 +24,7 @@ import (
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/extension/counter"
 	"github.com/uber/submitqueue/platform/metrics"
+	corerequest "github.com/uber/submitqueue/submitqueue/core/request"
 	"github.com/uber/submitqueue/submitqueue/core/topickey"
 	"github.com/uber/submitqueue/submitqueue/entity"
 	"github.com/uber/submitqueue/submitqueue/extension/conflict"
@@ -287,6 +288,20 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 		"dependency_count", len(batch.Dependencies),
 	)
 
+	// Record the "batched" status in the request log. This status corresponds to
+	// the RequestStateBatched transition CAS'd above, so it carries the request
+	// version for reconciliation (unlike the batch-level "scored" status). The
+	// message ID is scoped to (requestID, status), so a redelivery that creates a
+	// fresh batch re-emits "batched" with a different batch_id but is deduped to
+	// the first entry — acceptable, the request is batched either way.
+	logEntry := entity.NewRequestLog(request.ID, entity.RequestStatusBatched, request.Version, "", map[string]string{
+		"batch_id": batch.ID,
+	})
+	if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID); err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "request_log_errors", 1)
+		return fmt.Errorf("failed to publish request log for request %s: %w", request.ID, err)
+	}
+
 	// Publish to score topic for further processing.
 	// If it fails and the controller retries, a new batch will be created with the new batch ID but the same request ID.
 	// The downstream logic should be able to handle stale entries by looking at the state of the batch.

submitqueue/orchestrator/controller/batch/batch_test.go

@@ -75,7 +75,10 @@ func testRequest() entity.Request {
 // newTestController creates a controller with test dependencies.
 // If mockStorage is nil, a default MockStorage with an empty batch store is created.
 // If analyzer is nil, the "all" conflict analyzer is used (every active batch becomes a dependency).
-func newTestController(t *testing.T, ctrl *gomock.Controller, cnt *countermock.MockCounter, mockStorage *storagemock.MockStorage, analyzer conflict.Analyzer, publishErr error) *Controller {
+// scorePublishErr, if non-nil, is returned only for publishes to the "score" topic; the
+// log publish (which the controller emits first) always succeeds, so callers exercising the
+// score publish-failure path are not short-circuited on the earlier log publish.
+func newTestController(t *testing.T, ctrl *gomock.Controller, cnt *countermock.MockCounter, mockStorage *storagemock.MockStorage, analyzer conflict.Analyzer, scorePublishErr error) *Controller {
 	logger := zaptest.NewLogger(t).Sugar()
 	scope := tally.NoopScope
 
@@ -105,15 +108,21 @@ func newTestController(t *testing.T, ctrl *gomock.Controller, cnt *countermock.M
 	mockPub := queuemock.NewMockPublisher(ctrl)
 	mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
 		func(ctx context.Context, topic string, msg entityqueue.Message) error {
-			return publishErr
+			if topic == "score" {
+				return scorePublishErr
+			}
+			return nil
 		},
 	).AnyTimes()
 
 	mockQ := queuemock.NewMockQueue(ctrl)
 	mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes()
 
 	registry, err := consumer.NewTopicRegistry(
-		[]consumer.TopicConfig{{Key: topickey.TopicKeyScore, Name: "score", Queue: mockQ}},
+		[]consumer.TopicConfig{
+			{Key: topickey.TopicKeyScore, Name: "score", Queue: mockQ},
+			{Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ},
+		},
 	)
 	require.NoError(t, err)
 
@@ -148,6 +157,75 @@ func TestController_Process_Success(t *testing.T) {
 	require.NoError(t, err)
 }
 
+// TestController_Process_PublishesBatchedLog asserts the controller emits a
+// "batched" request log carrying the request ID, the post-CAS request version,
+// and the batch ID it was placed into.
+func TestController_Process_PublishesBatchedLog(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	request := testRequest()
+
+	mockBatchStore := storagemock.NewMockBatchStore(ctrl)
+	mockBatchStore.EXPECT().GetByQueueAndStates(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
+	mockBatchStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
+
+	mockReqStore := storagemock.NewMockRequestStore(ctrl)
+	mockReqStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil)
+	mockReqStore.EXPECT().UpdateState(gomock.Any(), request.ID, request.Version, request.Version+1, entity.RequestStateBatched).Return(nil)
+
+	mockBatchDependentStore := storagemock.NewMockBatchDependentStore(ctrl)
+	mockBatchDependentStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
+
+	mockStorage := storagemock.NewMockStorage(ctrl)
+	mockStorage.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes()
+	mockStorage.EXPECT().GetBatchDependentStore().Return(mockBatchDependentStore).AnyTimes()
+	mockStorage.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes()
+
+	// Capture messages published to the log topic.
+	var logMsgs []entityqueue.Message
+	mockPub := queuemock.NewMockPublisher(ctrl)
+	mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, topic string, msg entityqueue.Message) error {
+			if topic == "log" {
+				logMsgs = append(logMsgs, msg)
+			}
+			return nil
+		},
+	).AnyTimes()
+	mockQ := queuemock.NewMockQueue(ctrl)
+	mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes()
+
+	registry, err := consumer.NewTopicRegistry(
+		[]consumer.TopicConfig{
+			{Key: topickey.TopicKeyScore, Name: "score", Queue: mockQ},
+			{Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ},
+		},
+	)
+	require.NoError(t, err)
+
+	analyzerFactory := conflictmock.NewMockFactory(ctrl)
+	analyzerFactory.EXPECT().For(gomock.Any()).Return(all.New(), nil).AnyTimes()
+	controller := NewController(
+		zaptest.NewLogger(t).Sugar(), tally.NoopScope, registry, newSequentialCounter(ctrl),
+		mockStorage, analyzerFactory, topickey.TopicKeyBatch, "orchestrator-batch",
+	)
+
+	msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil)
+	delivery := queuemock.NewMockDelivery(ctrl)
+	delivery.EXPECT().Message().Return(msg).AnyTimes()
+	delivery.EXPECT().Attempt().Return(1).AnyTimes()
+
+	require.NoError(t, controller.Process(context.Background(), delivery))
+
+	require.Len(t, logMsgs, 1)
+	logEntry, err := entity.RequestLogFromBytes(logMsgs[0].Payload)
+	require.NoError(t, err)
+	assert.Equal(t, request.ID, logEntry.RequestID)
+	assert.Equal(t, entity.RequestStatusBatched, logEntry.Status)
+	assert.Equal(t, request.Version+1, logEntry.RequestVersion)
+	assert.Equal(t, "test-queue/batch/1", logEntry.Metadata["batch_id"])
+}
+
 func TestController_Process_StorageFailure(t *testing.T) {
 	ctrl := gomock.NewController(t)