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

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

@@ -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)
 	}
 

submitqueue/gateway/controller/cancel_test.go

@@ -28,7 +28,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"
@@ -57,14 +56,16 @@ 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
 }
 
 // testCancelRequest returns a valid entity.CancelRequest for testing.
@@ -75,14 +76,14 @@ func testCancelRequest(sqid string, reason string) entity.CancelRequest {
 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()
 
 	err := controller.Cancel(ctx, testCancelRequest("test-queue/42", "user changed their mind"))
@@ -93,7 +94,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()
 
 	err := controller.Cancel(ctx, testCancelRequest("", "anything"))
@@ -117,7 +118,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()
 
 	err := controller.Cancel(ctx, testCancelRequest("my-queue/7", "obsolete change"))
@@ -138,31 +139,28 @@ func TestCancel_PublishesToQueue(t *testing.T) {
 // before the cancel topic publish so observers see intent the moment Cancel returns.
 func TestCancel_InsertsCancellingLog(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 != ""
+			fixture.mu.Lock()
+			defer fixture.mu.Unlock()
+			insertedBeforePublish = len(fixture.logs) == 1
 			return nil
 		},
 	)
 
-	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, logStore, registry)
+	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, fixture.storage, registry)
 
 	err := controller.Cancel(context.Background(), testCancelRequest("my-queue/42", "obsolete change"))
 	require.NoError(t, err)
 
+	fixture.mu.Lock()
+	require.Len(t, fixture.logs, 1)
+	insertedLog := fixture.logs[0]
+	fixture.mu.Unlock()
 	assert.Equal(t, "my-queue/42", insertedLog.RequestID)
 	assert.Equal(t, entity.RequestStatusCancelling, insertedLog.Status)
 	assert.Equal(t, "obsolete change", insertedLog.Metadata["reason"])
@@ -173,15 +171,13 @@ func TestCancel_InsertsCancellingLog(t *testing.T) {
 // short-circuits the RPC with an error and the cancel topic is never published to.
 func TestCancel_LogInsertFailure(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)
 	_ = 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(), testCancelRequest("q/1", ""))
 	require.Error(t, err)
 }
@@ -192,27 +188,22 @@ func TestCancel_ReturnsErrorOnPublishFailure(t *testing.T) {
 	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)
+	controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorageFixture(ctrl, "test-queue/1").storage, registry)
 	ctx := context.Background()
 
 	err := controller.Cancel(ctx, testCancelRequest("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)
+	fixture := newCancelStorageFixture(ctrl, "")
 
 	registry, publisher := newCancelTestRegistry(t, ctrl)
 	_ = 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(), testCancelRequest("ghost/1", ""))
 	require.Error(t, err)
 	assert.True(t, IsRequestNotFound(err))
@@ -224,19 +215,21 @@ func TestCancel_UnknownSqidIsUserError(t *testing.T) {
 	assert.Equal(t, "ghost/1", typed.Sqid)
 }
 
-// TestCancel_RequestLogLookupFailure asserts that an infrastructure failure on
+// TestCancel_RequestSummaryLookupFailure 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)
+	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(), testCancelRequest("q/1", ""))
 	require.Error(t, err)
 	assert.False(t, errs.IsUserError(err))

submitqueue/gateway/controller/land.go

@@ -66,8 +66,8 @@ type LandController struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
 	counter       counter.Counter
-	store         storage.Storage
 	receiptWriter *requestcore.ReceiptWriter
+	materializer  *requestcore.Materializer
 	queueConfigs  queueconfig.Store
 	registry      consumer.TopicRegistry
 }
@@ -80,8 +80,8 @@ func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counter cou
 		logger:        logger,
 		metricsScope:  scope.SubScope("land_controller"),
 		counter:       counter,
-		store:         store,
 		receiptWriter: requestcore.NewReceiptWriter(store),
+		materializer:  requestcore.NewMaterializer(store),
 		queueConfigs:  queueConfigs,
 		registry:      registry,
 	}
@@ -148,7 +148,7 @@ func (c *LandController) Land(ctx context.Context, req entity.LandRequest) (resu
 		Status:      entity.RequestStatusAccepted,
 		Metadata:    map[string]string{},
 	}
-	if err := c.store.GetRequestLogStore().Insert(ctx, logEntry); err != nil {
+	if err := c.materializer.PersistLog(ctx, logEntry); err != nil {
 		// Publication is the Land success boundary. Returning an error here would
 		// encourage the caller to submit a duplicate request that is already queued.
 		c.logger.Errorw("failed to record accepted status after publishing request",