feat(submitqueue): activate request status materialization

Route synchronous, queued, and DLQ request-log writes through the shared materializer so authoritative and queue projections converge with retained history.

Preserve the original DLQ failure reason as the terminal request last error.

Validation: make fmt && make build && make test && make e2e-test

Author: albertywu

service/submitqueue/gateway/server/main.go

@@ -215,8 +215,9 @@ func run() error {
 	// Build the topic registry. The gateway publishes to the start of the
 	// orchestrator pipeline (TopicKeyStart) and the cancel topic (TopicKeyCancel) —
 	// both publish-only. It additionally consumes the log topic (TopicKeyLog):
-	// the gateway is the sole writer of the request log, persisting entries that
-	// the orchestrator publishes there.
+	// the gateway persists normal pipeline entries that the orchestrator
+	// publishes there. Orchestrator DLQ reconciliation may repair terminal
+	// request-log projections directly.
 	registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{
 		{Key: topickey.TopicKeyStart, Name: "start", Queue: mysqlQueue},
 		{Key: topickey.TopicKeyCancel, Name: "cancel", Queue: mysqlQueue},
@@ -249,8 +250,8 @@ func run() error {
 
 	// Initialize storage from the shared app database connection. The land
 	// controller writes to this store directly; cancel/status use the request
-	// log store directly. The log consumer (registered below) is the sole
-	// persister of request log entries published by the orchestrator.
+	// log store directly. The log consumer registered below persists request
+	// log entries published by the orchestrator's normal pipeline.
 	store, err := mysqlstorage.NewStorage(appDB, scope.SubScope("storage"))
 	if err != nil {
 		return fmt.Errorf("failed to create storage: %w", err)
@@ -271,7 +272,7 @@ func run() error {
 	// Create controllers and wrap them for gRPC
 	pingController := controller.NewPingController(logger, scope)
 	landController := controller.NewLandController(logger.Sugar(), scope, cnt, store, queueConfigs, registry)
-	cancelController := controller.NewCancelController(logger.Sugar(), scope, requestLogStore, registry)
+	cancelController := controller.NewCancelController(logger.Sugar(), scope, store, registry)
 	statusController := controller.NewStatusController(logger.Sugar(), scope, requestLogStore)
 	gatewayServer := &GatewayServer{
 		pingController:   pingController,
@@ -285,9 +286,10 @@ func run() error {
 	// Register reflection service for debugging with grpcurl
 	reflection.Register(grpcServer)
 
-	// Create the queue consumer and register the log controller. The gateway is
-	// the sole persister of the request log: the orchestrator publishes entries
-	// to the log topic and this consumer writes them to storage.
+	// Create the queue consumer and register the log controller. The orchestrator
+	// publishes normal pipeline entries to the log topic and this consumer writes
+	// them to storage. Orchestrator DLQ reconciliation repairs terminal entries
+	// directly so reconciliation does not depend on another asynchronous path.
 	logConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry,
 		errs.NewClassifierProcessor(
 			// Storage (submitqueue/extension/storage/mysql) and queue (platform/extension/messagequeue/mysql)

service/submitqueue/orchestrator/server/main.go

@@ -415,10 +415,9 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 		})
 	}
 
-	// Publish-only: the orchestrator emits request log entries to the log
-	// topic but never persists them. The gateway is the sole consumer that
-	// writes the request log to storage, so the orchestrator registers no
-	// consuming subscription (and therefore no log DLQ) for this topic.
+	// Publish-only: the orchestrator emits request-log entries to the log topic.
+	// The gateway is the sole consumer and writer of request logs and public
+	// projections, so the orchestrator registers no consuming subscription.
 	configs = append(configs, consumer.TopicConfig{
 		Key:   topickey.TopicKeyLog,
 		Name:  "log",
@@ -740,19 +739,19 @@ func registerDLQControllers(c consumer.Consumer, logger *zap.SugaredLogger, scop
 		name string
 		ctl  consumer.Controller
 	}{
-		{"start_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, dlq.DecodeLandRequestID, dlq.TopicKey(topickey.TopicKeyStart), "orchestrator-start-dlq")},
-		{"cancel_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, dlq.DecodeCancelRequestID, dlq.TopicKey(topickey.TopicKeyCancel), "orchestrator-cancel-dlq")},
-		{"validate_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, dlq.DecodeRequestID, dlq.TopicKey(topickey.TopicKeyValidate), "orchestrator-validate-dlq")},
-		{"mergeconflictsignal_dlq", dlq.NewDLQMergeConflictSignalController(logger, dlqScope, store, dlq.TopicKey(runwaymq.TopicKeyMergeConflictCheckSignal), "orchestrator-mergeconflictsignal-dlq")},
-		{"batch_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, dlq.DecodeRequestID, dlq.TopicKey(topickey.TopicKeyBatch), "orchestrator-batch-dlq")},
-		{"score_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeyScore), "orchestrator-score-dlq")},
-		{"speculate_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeySpeculate), "orchestrator-speculate-dlq")},
+		{"start_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, registry, dlq.DecodeLandRequestID, dlq.TopicKey(topickey.TopicKeyStart), "orchestrator-start-dlq")},
+		{"cancel_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, registry, dlq.DecodeCancelRequestID, dlq.TopicKey(topickey.TopicKeyCancel), "orchestrator-cancel-dlq")},
+		{"validate_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, registry, dlq.DecodeRequestID, dlq.TopicKey(topickey.TopicKeyValidate), "orchestrator-validate-dlq")},
+		{"mergeconflictsignal_dlq", dlq.NewDLQMergeConflictSignalController(logger, dlqScope, store, registry, dlq.TopicKey(runwaymq.TopicKeyMergeConflictCheckSignal), "orchestrator-mergeconflictsignal-dlq")},
+		{"batch_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, registry, dlq.DecodeRequestID, dlq.TopicKey(topickey.TopicKeyBatch), "orchestrator-batch-dlq")},
+		{"score_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyScore), "orchestrator-score-dlq")},
+		{"speculate_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeySpeculate), "orchestrator-speculate-dlq")},
 		{"prioritize_dlq", dlq.NewDLQQueueController(logger, dlqScope, registry, dlq.TopicKey(topickey.TopicKeyPrioritize), "orchestrator-prioritize-dlq")},
-		{"build_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeyBuild), "orchestrator-build-dlq")},
-		{"buildsignal_dlq", dlq.NewDLQBuildSignalController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq")},
-		{"merge_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq")},
-		{"mergesignal_dlq", dlq.NewDLQMergeSignalController(logger, dlqScope, store, dlq.TopicKey(runwaymq.TopicKeyMergeSignal), "orchestrator-mergesignal-dlq")},
-		{"conclude_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, dlq.TopicKey(topickey.TopicKeyConclude), "orchestrator-conclude-dlq")},
+		{"build_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyBuild), "orchestrator-build-dlq")},
+		{"buildsignal_dlq", dlq.NewDLQBuildSignalController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq")},
+		{"merge_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq")},
+		{"mergesignal_dlq", dlq.NewDLQMergeSignalController(logger, dlqScope, store, registry, dlq.TopicKey(runwaymq.TopicKeyMergeSignal), "orchestrator-mergesignal-dlq")},
+		{"conclude_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyConclude), "orchestrator-conclude-dlq")},
 	}
 	var count int
 	for _, reg := range dlqRegs {

submitqueue/gateway/controller/cancel.go

@@ -23,6 +23,7 @@ import (
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/errs"
+	requestcore "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/storage"
@@ -35,21 +36,23 @@ import (
 // and returns a response. The orchestrator-side cancel controller performs the actual
 // state transitions and emits the terminal RequestStatusCancelled log entry.
 type CancelController struct {
-	logger          *zap.SugaredLogger
-	metricsScope    tally.Scope
-	requestLogStore storage.RequestLogStore
-	registry        consumer.TopicRegistry
+	logger              *zap.SugaredLogger
+	metricsScope        tally.Scope
+	requestSummaryStore storage.RequestSummaryStore
+	materializer        *requestcore.Materializer
+	registry            consumer.TopicRegistry
 }
 
 // NewCancelController creates a new instance of the gateway cancel controller.
-// The controller writes a RequestStatusCancelling log entry through requestLogStore and
+// The controller writes a RequestStatusCancelling log entry through the shared materializer and
 // publishes cancel requests to the topic registered under topickey.TopicKeyCancel.
-func NewCancelController(logger *zap.SugaredLogger, scope tally.Scope, requestLogStore storage.RequestLogStore, registry consumer.TopicRegistry) *CancelController {
+func NewCancelController(logger *zap.SugaredLogger, scope tally.Scope, store storage.Storage, registry consumer.TopicRegistry) *CancelController {
 	return &CancelController{
-		logger:          logger,
-		metricsScope:    scope,
-		requestLogStore: requestLogStore,
-		registry:        registry,
+		logger:              logger,
+		metricsScope:        scope,
+		requestSummaryStore: store.GetRequestSummaryStore(),
+		materializer:        requestcore.NewMaterializer(store),
+		registry:            registry,
 	}
 }
 
@@ -78,18 +81,13 @@ func (c *CancelController) Cancel(ctx context.Context, req entity.CancelRequest)
 		"reason", req.Reason,
 	)
 
-	// Verify the sqid exists before recording intent or publishing. Cancel is opt-in
-	// by sqid; an unknown sqid is a user error and must never leave a cancelling log
-	// row or a queue message behind for a request that never existed. The Land
-	// controller writes its "accepted" log entry synchronously to the same store, so
-	// a NotFound here reliably means "this sqid was never accepted by the gateway"
-	// rather than "in flight" — there is no false-negative race window.
-	if _, err := c.requestLogStore.List(ctx, req.ID); err != nil {
+	// Verify the sqid exists before recording intent or publishing.
+	if _, err := c.requestSummaryStore.Get(ctx, req.ID); err != nil {
 		if storage.IsNotFound(err) {
 			c.metricsScope.Counter("cancel_request_not_found").Inc(1)
 			return errs.NewUserError(&RequestNotFoundError{Sqid: req.ID})
 		}
-		return fmt.Errorf("CancelController failed to look up request log for sqid=%s: %w", req.ID, err)
+		return fmt.Errorf("CancelController failed to look up request summary for sqid=%s: %w", req.ID, err)
 	}
 
 	// Record the user's intent in the request log before publishing. Writing direct to the
@@ -100,7 +98,7 @@ func (c *CancelController) Cancel(ctx context.Context, req entity.CancelRequest)
 		metadata["reason"] = req.Reason
 	}
 	logEntry := entity.NewRequestLog(req.ID, entity.RequestStatusCancelling, 0, "", metadata)
-	if err := c.requestLogStore.Insert(ctx, logEntry); err != nil {
+	if err := c.materializer.PersistLog(ctx, logEntry); err != nil {
 		return fmt.Errorf("CancelController failed to insert cancelling log for sqid=%s: %w", req.ID, err)
 	}