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

@@ -252,7 +252,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,

submitqueue/gateway/controller/cancel.go

@@ -24,6 +24,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"
@@ -36,21 +37,22 @@ 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
-// 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 {
+// 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, 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,
 	}
 }
 
@@ -84,18 +86,13 @@ func (c *CancelController) Cancel(ctx context.Context, req *pb.CancelRequest) (*
 		"reason", cancelRequest.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, cancelRequest.ID); err != nil {
+	// 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.
+	if _, err := c.requestSummaryStore.Get(ctx, cancelRequest.ID); err != nil {
 		if storage.IsNotFound(err) {
 			c.metricsScope.Counter("cancel_request_not_found").Inc(1)
 			return nil, errs.NewUserError(&RequestNotFoundError{Sqid: cancelRequest.ID})
 		}
-		return nil, fmt.Errorf("CancelController failed to look up request log for sqid=%s: %w", cancelRequest.ID, err)
+		return nil, fmt.Errorf("CancelController failed to look up request summary for sqid=%s: %w", cancelRequest.ID, err)
 	}
 
 	// Record the user's intent in the request log before publishing. Writing direct to the
@@ -106,7 +103,7 @@ func (c *CancelController) Cancel(ctx context.Context, req *pb.CancelRequest) (*
 		metadata["reason"] = cancelRequest.Reason
 	}
 	logEntry := entity.NewRequestLog(cancelRequest.ID, entity.RequestStatusCancelling, 0, "", metadata)
-	if err := c.requestLogStore.Insert(ctx, logEntry); err != nil {
+	if err := c.materializer.PersistLog(ctx, logEntry); err != nil {
 		return nil, fmt.Errorf("CancelController failed to insert cancelling log for sqid=%s: %w", cancelRequest.ID, err)
 	}
 

submitqueue/gateway/controller/cancel_test.go

@@ -29,7 +29,6 @@ import (
 	queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
 	"github.com/uber/submitqueue/submitqueue/core/topickey"
 	"github.com/uber/submitqueue/submitqueue/entity"
-	"github.com/uber/submitqueue/submitqueue/extension/storage"
 	storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock"
 	"go.uber.org/mock/gomock"
 	"go.uber.org/zap"
@@ -58,27 +57,29 @@ func newCancelTestRegistryWithNoopPublisher(t *testing.T, ctrl *gomock.Controlle
 	return registry
 }
 
-// newRequestLogStoreNoop returns a RequestLogStore mock whose List returns a single
-// dummy entry (so existence check passes) and whose Insert silently succeeds for any input.
-func newRequestLogStoreNoop(t *testing.T, ctrl *gomock.Controller) *storagemock.MockRequestLogStore {
-	t.Helper()
-	store := storagemock.NewMockRequestLogStore(ctrl)
-	store.EXPECT().List(gomock.Any(), gomock.Any()).Return([]entity.RequestLog{{}}, nil).AnyTimes()
-	store.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
-	return store
+// newCancelStorageFixture returns a storage fixture with one received request.
+func newCancelStorageFixture(ctrl *gomock.Controller, requestID string) *controllerStorageFixture {
+	fixture := newControllerStorageFixture(ctrl)
+	if requestID != "" {
+		fixture.addSummary(entity.RequestSummary{
+			RequestID: requestID, Queue: "test-queue", ChangeURIs: []string{}, ReceivedAtMs: 1,
+			Status: entity.RequestStatusAccepted, StatusTimestampMs: 1, Version: 1, Metadata: map[string]string{},
+		})
+	}
+	return fixture
 }
 
 func TestNewCancelController(t *testing.T) {
 	ctrl := gomock.NewController(t)
 
-	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newRequestLogStoreNoop(t, ctrl), newCancelTestRegistryWithNoopPublisher(t, ctrl))
+	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorageFixture(ctrl, "test-queue/42").storage, newCancelTestRegistryWithNoopPublisher(t, ctrl))
 	require.NotNil(t, controller)
 }
 
 func TestCancel_HappyPath(t *testing.T) {
 	ctrl := gomock.NewController(t)
 
-	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newRequestLogStoreNoop(t, ctrl), newCancelTestRegistryWithNoopPublisher(t, ctrl))
+	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorageFixture(ctrl, "test-queue/42").storage, newCancelTestRegistryWithNoopPublisher(t, ctrl))
 	ctx := context.Background()
 
 	req := &pb.CancelRequest{Sqid: "test-queue/42", Reason: "user changed their mind"}
@@ -91,7 +92,7 @@ func TestCancel_HappyPath(t *testing.T) {
 func TestCancel_ReturnsErrorOnEmptySqid(t *testing.T) {
 	ctrl := gomock.NewController(t)
 
-	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newRequestLogStoreNoop(t, ctrl), newCancelTestRegistryWithNoopPublisher(t, ctrl))
+	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorageFixture(ctrl, "test-queue/42").storage, newCancelTestRegistryWithNoopPublisher(t, ctrl))
 	ctx := context.Background()
 
 	req := &pb.CancelRequest{Sqid: "", Reason: "anything"}
@@ -116,7 +117,7 @@ func TestCancel_PublishesToQueue(t *testing.T) {
 		},
 	)
 
-	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newRequestLogStoreNoop(t, ctrl), registry)
+	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorageFixture(ctrl, "my-queue/7").storage, registry)
 	ctx := context.Background()
 
 	req := &pb.CancelRequest{Sqid: "my-queue/7", Reason: "obsolete change"}
@@ -133,91 +134,57 @@ func TestCancel_PublishesToQueue(t *testing.T) {
 	assert.Equal(t, "obsolete change", deserialized.Reason)
 }
 
-// TestCancel_InsertsCancellingLog asserts that Cancel records a RequestStatusCancelling
-// log entry (intent) carrying the reason in metadata, and that the entry is written
-// before the cancel topic publish so observers see intent the moment Cancel returns.
-func TestCancel_InsertsCancellingLog(t *testing.T) {
+// TestCancel_MaterializesCancellingBeforePublish asserts that Cancel records the intent before publishing.
+func TestCancel_MaterializesCancellingBeforePublish(t *testing.T) {
 	ctrl := gomock.NewController(t)
-
-	var insertedLog entity.RequestLog
-	logStore := storagemock.NewMockRequestLogStore(ctrl)
-	logStore.EXPECT().List(gomock.Any(), "my-queue/42").Return([]entity.RequestLog{{}}, nil)
-	logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).DoAndReturn(
-		func(_ context.Context, entry entity.RequestLog) error {
-			insertedLog = entry
-			return nil
-		},
-	).Times(1)
+	fixture := newCancelStorageFixture(ctrl, "my-queue/42")
 
 	registry, publisher := newCancelTestRegistry(t, ctrl)
-	insertedBeforePublish := false
 	publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
 		func(_ context.Context, _ string, _ entityqueue.Message) error {
-			insertedBeforePublish = insertedLog.RequestID != ""
+			summary, err := fixture.summaryStore.Get(context.Background(), "my-queue/42")
+			require.NoError(t, err)
+			assert.Equal(t, entity.RequestStatusCancelling, summary.Status)
+			assert.Equal(t, "obsolete change", summary.Metadata["reason"])
 			return nil
 		},
 	)
 
-	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, logStore, registry)
-
-	req := &pb.CancelRequest{Sqid: "my-queue/42", Reason: "obsolete change"}
-	_, err := controller.Cancel(context.Background(), req)
+	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, fixture.storage, registry)
+	_, err := controller.Cancel(context.Background(), &pb.CancelRequest{Sqid: "my-queue/42", Reason: "obsolete change"})
 	require.NoError(t, err)
-
-	assert.Equal(t, "my-queue/42", insertedLog.RequestID)
-	assert.Equal(t, entity.RequestStatusCancelling, insertedLog.Status)
-	assert.Equal(t, "obsolete change", insertedLog.Metadata["reason"])
-	assert.True(t, insertedBeforePublish, "log entry must be inserted before publish to the cancel topic")
 }
 
-// TestCancel_LogInsertFailure asserts that a failure to insert the Cancelling log entry
-// short-circuits the RPC with an error and the cancel topic is never published to.
-func TestCancel_LogInsertFailure(t *testing.T) {
+func TestCancel_LogInsertFailureSkipsPublish(t *testing.T) {
 	ctrl := gomock.NewController(t)
-
-	logStore := storagemock.NewMockRequestLogStore(ctrl)
-	logStore.EXPECT().List(gomock.Any(), "q/1").Return([]entity.RequestLog{{}}, nil)
-	logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(fmt.Errorf("db unavailable"))
-
+	fixture := newCancelStorageFixture(ctrl, "q/1")
+	fixture.setLogInsertError(fmt.Errorf("db unavailable"))
 	registry, publisher := newCancelTestRegistry(t, ctrl)
-	// No Publish expectation: log insert must fail before publish runs.
 	_ = publisher
 
-	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, logStore, registry)
+	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, fixture.storage, registry)
 	_, err := controller.Cancel(context.Background(), &pb.CancelRequest{Sqid: "q/1"})
 	require.Error(t, err)
 }
 
 func TestCancel_ReturnsErrorOnPublishFailure(t *testing.T) {
 	ctrl := gomock.NewController(t)
-
+	fixture := newCancelStorageFixture(ctrl, "test-queue/1")
 	registry, publisher := newCancelTestRegistry(t, ctrl)
 	publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("queue unavailable"))
 
-	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newRequestLogStoreNoop(t, ctrl), registry)
-	ctx := context.Background()
-
-	req := &pb.CancelRequest{Sqid: "test-queue/1"}
-	_, err := controller.Cancel(ctx, req)
-
+	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, fixture.storage, registry)
+	_, err := controller.Cancel(context.Background(), &pb.CancelRequest{Sqid: "test-queue/1"})
 	require.Error(t, err)
 }
 
-// TestCancel_UnknownSqidIsUserError asserts that Cancel for a sqid with no
-// request_log history fails fast with a RequestNotFoundError (user error) and
-// never inserts a cancelling log row or publishes to the cancel topic.
 func TestCancel_UnknownSqidIsUserError(t *testing.T) {
 	ctrl := gomock.NewController(t)
-
-	logStore := storagemock.NewMockRequestLogStore(ctrl)
-	logStore.EXPECT().List(gomock.Any(), "ghost/1").Return(nil, storage.ErrNotFound)
-	// No Insert expectation: existence check must short-circuit before Insert.
-
+	fixture := newCancelStorageFixture(ctrl, "")
 	registry, publisher := newCancelTestRegistry(t, ctrl)
-	// No Publish expectation: existence check must short-circuit before Publish.
 	_ = publisher
 
-	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, logStore, registry)
+	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, fixture.storage, registry)
 	_, err := controller.Cancel(context.Background(), &pb.CancelRequest{Sqid: "ghost/1"})
 	require.Error(t, err)
 	assert.True(t, IsRequestNotFound(err))
@@ -229,19 +196,19 @@ func TestCancel_UnknownSqidIsUserError(t *testing.T) {
 	assert.Equal(t, "ghost/1", typed.Sqid)
 }
 
-// TestCancel_RequestLogLookupFailure asserts that an infrastructure failure on
-// the existence check propagates as a (non-user) error and skips the rest of
-// the pipeline.
-func TestCancel_RequestLogLookupFailure(t *testing.T) {
+func TestCancel_RequestSummaryLookupFailure(t *testing.T) {
 	ctrl := gomock.NewController(t)
-
-	logStore := storagemock.NewMockRequestLogStore(ctrl)
-	logStore.EXPECT().List(gomock.Any(), "q/1").Return(nil, fmt.Errorf("log backend down"))
-
+	store := storagemock.NewMockStorage(ctrl)
+	summaryStore := storagemock.NewMockRequestSummaryStore(ctrl)
+	store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes()
+	store.EXPECT().GetRequestLogStore().Return(storagemock.NewMockRequestLogStore(ctrl)).AnyTimes()
+	store.EXPECT().GetRequestQueueSummaryStore().Return(storagemock.NewMockRequestQueueSummaryStore(ctrl)).AnyTimes()
+	store.EXPECT().GetRequestURIStore().Return(storagemock.NewMockRequestURIStore(ctrl)).AnyTimes()
+	summaryStore.EXPECT().Get(gomock.Any(), "q/1").Return(entity.RequestSummary{}, fmt.Errorf("summary backend down"))
 	registry, publisher := newCancelTestRegistry(t, ctrl)
 	_ = publisher
 
-	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, logStore, registry)
+	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, store, registry)
 	_, err := controller.Cancel(context.Background(), &pb.CancelRequest{Sqid: "q/1"})
 	require.Error(t, err)
 	assert.False(t, errs.IsUserError(err))

submitqueue/gateway/controller/land.go

@@ -72,8 +72,8 @@ type LandController struct {
 	logger          *zap.SugaredLogger
 	metricsScope    tally.Scope
 	counter         counter.Counter
-	store           storage.Storage
 	admissionWriter *requestcore.AdmissionWriter
+	materializer    *requestcore.Materializer
 	queueConfigs    queueconfig.Store
 	registry        consumer.TopicRegistry
 }
@@ -86,8 +86,8 @@ func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counter cou
 		logger:          logger,
 		metricsScope:    scope.SubScope("land_controller"),
 		counter:         counter,
-		store:           store,
 		admissionWriter: requestcore.NewAdmissionWriter(store),
+		materializer:    requestcore.NewMaterializer(store),
 		queueConfigs:    queueConfigs,
 		registry:        registry,
 	}
@@ -168,8 +168,8 @@ func (c *LandController) Land(ctx context.Context, req *pb.LandRequest) (resp *p
 		Status:      entity.RequestStatusAccepted,
 		Metadata:    map[string]string{},
 	}
-	if err := c.store.GetRequestLogStore().Insert(ctx, logEntry); err != nil {
-		return nil, fmt.Errorf("LandController failed to insert request log for sqid=%s: %w", landRequest.ID, err)
+	if err := c.materializer.PersistLog(ctx, logEntry); err != nil {
+		return nil, fmt.Errorf("LandController failed to persist accepted status for sqid=%s: %w", landRequest.ID, err)
 	}
 
 	c.logger.Debugw("land request created",

submitqueue/gateway/controller/land_test.go

@@ -330,6 +330,16 @@ func TestLand_PublishesToQueue(t *testing.T) {
 				return nil
 			},
 		),
+		summaryStore.EXPECT().Get(gomock.Any(), "test-queue/123").DoAndReturn(
+			func(context.Context, string) (entity.RequestSummary, error) {
+				return persistedSummary, nil
+			},
+		),
+		queueStore.EXPECT().Get(gomock.Any(), "test-queue", gomock.Any(), "test-queue/123").DoAndReturn(
+			func(context.Context, string, int64, string) (entity.RequestQueueSummary, error) {
+				return persistedQueueSummary, nil
+			},
+		),
 		publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
 			func(ctx context.Context, topic string, msg entityqueue.Message) error {
 				publishedTopic = topic

submitqueue/gateway/controller/log/BUILD.bazel

@@ -8,6 +8,7 @@ go_library(
     deps = [
         "//platform/consumer:go_default_library",
         "//platform/metrics:go_default_library",
+        "//submitqueue/core/request:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",

submitqueue/gateway/controller/log/log.go

@@ -21,6 +21,7 @@ import (
 	"github.com/uber-go/tally"
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/metrics"
+	requestcore "github.com/uber/submitqueue/submitqueue/core/request"
 	"github.com/uber/submitqueue/submitqueue/entity"
 	"github.com/uber/submitqueue/submitqueue/extension/storage"
 	"go.uber.org/zap"
@@ -36,7 +37,7 @@ import (
 type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
-	store         storage.Storage
+	materializer  *requestcore.Materializer
 	topicKey      consumer.TopicKey
 	consumerGroup string
 }
@@ -55,7 +56,7 @@ func NewController(
 	return &Controller{
 		logger:        logger.Named("log_controller"),
 		metricsScope:  scope.SubScope("log_controller"),
-		store:         store,
+		materializer:  requestcore.NewMaterializer(store),
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
 	}
@@ -87,10 +88,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 		"attempt", delivery.Attempt(),
 	)
 
-	// Persist request log to storage
-	if err := c.store.GetRequestLogStore().Insert(ctx, logEntry); err != nil {
+	// Persist the audit log and materialized views as one retryable application operation.
+	if err := c.materializer.PersistLog(ctx, logEntry); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
-		return fmt.Errorf("failed to insert request log: %w", err)
+		return fmt.Errorf("failed to persist request log: %w", err)
 	}
 
 	return nil // Success - message will be acked