diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index 73cd122d..427c76dd 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/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) diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 4a398add..eb1ce9ab 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/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 { diff --git a/submitqueue/gateway/controller/cancel.go b/submitqueue/gateway/controller/cancel.go index fa44d7c8..298345b1 100644 --- a/submitqueue/gateway/controller/cancel.go +++ b/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) } diff --git a/submitqueue/gateway/controller/cancel_test.go b/submitqueue/gateway/controller/cancel_test.go index 85b0662a..1e4161ca 100644 --- a/submitqueue/gateway/controller/cancel_test.go +++ b/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,7 +188,7 @@ 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", "")) @@ -200,19 +196,14 @@ func TestCancel_ReturnsErrorOnPublishFailure(t *testing.T) { 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)) diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go index 400f1ce6..6175fcc0 100644 --- a/submitqueue/gateway/controller/land.go +++ b/submitqueue/gateway/controller/land.go @@ -26,6 +26,7 @@ import ( "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/extension/counter" "github.com/uber/submitqueue/platform/metrics" + 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/queueconfig" @@ -66,6 +67,7 @@ type LandController struct { metricsScope tally.Scope counter counter.Counter store storage.Storage + materializer *requestcore.Materializer queueConfigs queueconfig.Store registry consumer.TopicRegistry } @@ -79,6 +81,7 @@ func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counter cou metricsScope: scope.SubScope("land_controller"), counter: counter, store: store, + materializer: requestcore.NewMaterializer(store), queueConfigs: queueConfigs, registry: registry, } @@ -145,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. Later pipeline events repair // the accepting projection even if this accepted log is not persisted. If // the client retries after losing the response, the orchestrator rejects diff --git a/submitqueue/gateway/controller/land_test.go b/submitqueue/gateway/controller/land_test.go index 109fbca7..80fcc081 100644 --- a/submitqueue/gateway/controller/land_test.go +++ b/submitqueue/gateway/controller/land_test.go @@ -309,7 +309,10 @@ func TestLand_PropagatesQueueConfigStoreError(t *testing.T) { func TestLand_PublishesToQueue(t *testing.T) { var publishedTopic string var publishedMessage entityqueue.Message - var persistedSummary entity.RequestSummary + var receiptSummary entity.RequestSummary + var materializedSummary entity.RequestSummary + var persistedMapping entity.RequestURI + var persistedQueueSummary entity.RequestQueueSummary var persistedLog entity.RequestLog ctrl := gomock.NewController(t) @@ -319,15 +322,19 @@ func TestLand_PublishesToQueue(t *testing.T) { store := storagemock.NewMockStorage(ctrl) summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + uriStore := storagemock.NewMockRequestURIStore(ctrl) + queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl) logStore := storagemock.NewMockRequestLogStore(ctrl) store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() + store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes() + store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes() store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() registry, publisher := newTestRegistry(t, ctrl) gomock.InOrder( summaryStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( func(_ context.Context, summary entity.RequestSummary) error { - persistedSummary = summary + receiptSummary = summary return nil }, ), @@ -344,6 +351,35 @@ 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 receiptSummary, nil + }, + ), + summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).DoAndReturn( + func(_ context.Context, summary entity.RequestSummary, _, newVersion int32) error { + summary.Version = newVersion + materializedSummary = summary + return nil + }, + ), + queueStore.EXPECT().Get(gomock.Any(), "test-queue", gomock.Any(), "test-queue/123").DoAndReturn( + func(context.Context, string, int64, string) (entity.RequestQueueSummary, error) { + return entity.RequestQueueSummary{}, storage.ErrNotFound + }, + ), + uriStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, mapping entity.RequestURI) error { + persistedMapping = mapping + return nil + }, + ), + queueStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, summary entity.RequestQueueSummary) error { + persistedQueueSummary = summary + return nil + }, + ), ) controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, cnt, store, noopQueueConfigStore(ctrl), registry) @@ -363,19 +399,35 @@ func TestLand_PublishesToQueue(t *testing.T) { RequestID: "test-queue/123", Queue: "test-queue", ChangeURIs: []string{"github://github.example.com/uber/backend/pull/456/fedcba9876543210fedcba9876543210fedcba98"}, - ReceivedAtMs: persistedSummary.ReceivedAtMs, + ReceivedAtMs: receiptSummary.ReceivedAtMs, Status: entity.RequestStatusAccepting, - StatusTimestampMs: persistedSummary.ReceivedAtMs, + StatusTimestampMs: receiptSummary.ReceivedAtMs, Version: 1, Metadata: map[string]string{}, - }, persistedSummary) - assert.Positive(t, persistedSummary.ReceivedAtMs) + }, receiptSummary) + assert.Positive(t, receiptSummary.ReceivedAtMs) assert.Equal(t, entity.RequestLog{ RequestID: "test-queue/123", - TimestampMs: persistedSummary.ReceivedAtMs, + TimestampMs: receiptSummary.ReceivedAtMs, Status: entity.RequestStatusAccepted, Metadata: map[string]string{}, }, persistedLog) + assert.Equal(t, entity.RequestStatusAccepted, materializedSummary.Status) + assert.Equal(t, int32(2), materializedSummary.Version) + assert.Equal(t, entity.RequestURI{ + ChangeURI: "github://github.example.com/uber/backend/pull/456/fedcba9876543210fedcba9876543210fedcba98", + ReceivedAtMs: receiptSummary.ReceivedAtMs, + RequestID: "test-queue/123", + }, persistedMapping) + assert.Equal(t, entity.RequestQueueSummary{ + RequestID: "test-queue/123", + Queue: "test-queue", + ChangeURIs: []string{"github://github.example.com/uber/backend/pull/456/fedcba9876543210fedcba9876543210fedcba98"}, + ReceivedAtMs: receiptSummary.ReceivedAtMs, + Status: entity.RequestStatusAccepted, + Version: 2, + Metadata: map[string]string{}, + }, persistedQueueSummary) // Verify message was published to the topic registered under TopicKeyStart assert.Equal(t, "start", publishedTopic) diff --git a/submitqueue/gateway/controller/log/BUILD.bazel b/submitqueue/gateway/controller/log/BUILD.bazel index 68608a82..646a5c1b 100644 --- a/submitqueue/gateway/controller/log/BUILD.bazel +++ b/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", diff --git a/submitqueue/gateway/controller/log/log.go b/submitqueue/gateway/controller/log/log.go index d28811c7..730add0d 100644 --- a/submitqueue/gateway/controller/log/log.go +++ b/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 diff --git a/submitqueue/gateway/controller/log/log_test.go b/submitqueue/gateway/controller/log/log_test.go index 166352bb..d5c4019c 100644 --- a/submitqueue/gateway/controller/log/log_test.go +++ b/submitqueue/gateway/controller/log/log_test.go @@ -30,55 +30,56 @@ import ( "go.uber.org/zap/zaptest" ) -// newTestController creates a controller with test dependencies. -func newTestController(t *testing.T, ctrl *gomock.Controller, store *storagemock.MockStorage) *Controller { - logger := zaptest.NewLogger(t).Sugar() - scope := tally.NoopScope - - return NewController(logger, scope, store, topickey.TopicKeyLog, "gateway-log") -} - func TestController_Process(t *testing.T) { tests := []struct { name string - logEntry *entity.RequestLog // nil means use rawPayload instead - rawPayload []byte // used when logEntry is nil (e.g. invalid JSON) + logEntry *entity.RequestLog + rawPayload []byte setupStore func(*gomock.Controller) *storagemock.MockStorage wantErr bool }{ { - name: "success", - logEntry: newRequestLog( - "test-queue/1", entity.RequestStatusStarted, 1, "", nil, - ), + name: "success", + logEntry: newRequestLog("test-queue/1", entity.RequestStatusStarted, 1, "", nil), setupStore: func(ctrl *gomock.Controller) *storagemock.MockStorage { - mockLogStore := storagemock.NewMockRequestLogStore(ctrl) - mockLogStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil) - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetRequestLogStore().Return(mockLogStore).AnyTimes() - return store + return newLogControllerStore(ctrl, nil, nil, nil, nil) }, - wantErr: false, }, { name: "invalid JSON", rawPayload: []byte(`{"invalid": json"}`), + setupStore: func(ctrl *gomock.Controller) *storagemock.MockStorage { return storagemock.NewMockStorage(ctrl) }, + wantErr: true, + }, + { + name: "audit insert failure", + logEntry: newRequestLog("test-queue/2", entity.RequestStatusError, 3, "merge conflict", nil), setupStore: func(ctrl *gomock.Controller) *storagemock.MockStorage { - return storagemock.NewMockStorage(ctrl) + return newLogControllerStore(ctrl, fmt.Errorf("audit down"), nil, nil, nil) }, wantErr: true, }, { - name: "storage failure", - logEntry: newRequestLog( - "test-queue/2", entity.RequestStatusError, 3, "merge conflict", map[string]string{"step": "merge"}, - ), + name: "summary read failure", + logEntry: newRequestLog("test-queue/2", entity.RequestStatusError, 3, "merge conflict", nil), setupStore: func(ctrl *gomock.Controller) *storagemock.MockStorage { - mockLogStore := storagemock.NewMockRequestLogStore(ctrl) - mockLogStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(fmt.Errorf("database connection failed")) - store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetRequestLogStore().Return(mockLogStore).AnyTimes() - return store + return newLogControllerStore(ctrl, nil, fmt.Errorf("summary down"), nil, nil) + }, + wantErr: true, + }, + { + name: "summary update failure", + logEntry: newRequestLog("test-queue/2", entity.RequestStatusError, 3, "merge conflict", nil), + setupStore: func(ctrl *gomock.Controller) *storagemock.MockStorage { + return newLogControllerStore(ctrl, nil, nil, fmt.Errorf("summary update down"), nil) + }, + wantErr: true, + }, + { + name: "queue projection failure", + logEntry: newRequestLog("test-queue/2", entity.RequestStatusError, 3, "merge conflict", nil), + setupStore: func(ctrl *gomock.Controller) *storagemock.MockStorage { + return newLogControllerStore(ctrl, nil, nil, nil, fmt.Errorf("queue update down")) }, wantErr: true, }, @@ -87,26 +88,19 @@ func TestController_Process(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { ctrl := gomock.NewController(t) - - var payload []byte + payload := tt.rawPayload if tt.logEntry != nil { var err error payload, err = tt.logEntry.ToBytes() require.NoError(t, err) - } else { - payload = tt.rawPayload } - - store := tt.setupStore(ctrl) - controller := newTestController(t, ctrl, store) - + controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, tt.setupStore(ctrl), topickey.TopicKeyLog, "gateway-log") msg := entityqueue.NewMessage("test-queue/1", payload, "test-queue", nil) delivery := queuemock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() err := controller.Process(context.Background(), delivery) - if tt.wantErr { require.Error(t, err) } else { @@ -116,7 +110,37 @@ func TestController_Process(t *testing.T) { } } -// newRequestLog is a helper that returns a pointer to a RequestLog for use in test tables. +func newLogControllerStore(ctrl *gomock.Controller, insertErr, getErr, updateErr, queueErr error) *storagemock.MockStorage { + store := storagemock.NewMockStorage(ctrl) + logStore := storagemock.NewMockRequestLogStore(ctrl) + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl) + store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() + store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() + store.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes() + logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(insertErr) + if insertErr != nil { + return store + } + summaryStore.EXPECT().Get(gomock.Any(), gomock.Any()).Return(entity.RequestSummary{ + RequestID: "test-queue/2", Queue: "test-queue", ChangeURIs: []string{}, ReceivedAtMs: 1, + Status: entity.RequestStatusAccepted, StatusTimestampMs: 1, Version: 1, Metadata: map[string]string{}, + }, getErr) + if getErr != nil { + return store + } + summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(updateErr) + if updateErr != nil { + return store + } + queueStore.EXPECT().Get(gomock.Any(), "test-queue", int64(1), "test-queue/2").Return(entity.RequestQueueSummary{ + RequestID: "test-queue/2", Queue: "test-queue", ChangeURIs: []string{}, ReceivedAtMs: 1, + Status: entity.RequestStatusAccepted, Version: 1, Metadata: map[string]string{}, + }, nil) + queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(queueErr) + return store +} + func newRequestLog(requestID string, status entity.RequestStatus, requestVersion int32, lastError string, metadata map[string]string) *entity.RequestLog { log := entity.NewRequestLog(requestID, status, requestVersion, lastError, metadata) return &log diff --git a/submitqueue/orchestrator/controller/dlq/BUILD.bazel b/submitqueue/orchestrator/controller/dlq/BUILD.bazel index d6aea342..f36148b9 100644 --- a/submitqueue/orchestrator/controller/dlq/BUILD.bazel +++ b/submitqueue/orchestrator/controller/dlq/BUILD.bazel @@ -19,6 +19,7 @@ go_library( "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", + "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage:go_default_library", @@ -36,6 +37,7 @@ go_test( "log_test.go", "mergeconflictsignal_test.go", "mergesignal_test.go", + "publisher_test.go", "queue_test.go", "request_test.go", ], diff --git a/submitqueue/orchestrator/controller/dlq/README.md b/submitqueue/orchestrator/controller/dlq/README.md index 3e12f432..a0a028f3 100644 --- a/submitqueue/orchestrator/controller/dlq/README.md +++ b/submitqueue/orchestrator/controller/dlq/README.md @@ -20,9 +20,9 @@ The recognised error condition is handled explicitly in `dlq.go`: Everything else — including `storage.ErrVersionMismatch` on the CAS — is returned plain and, after the always-retryable processor wrap, redelivered until it either succeeds or hits the attempt cap. There is no point in pre-classifying retryability at this layer when the processor forces every non-nil error retryable anyway. -## Request log entries are written directly, not via the queue +## Request log entries are published to Gateway -When a DLQ controller transitions a request to `RequestStateError` it also appends a `RequestStatusError` row to `RequestLogStore` via a direct `Insert` call, rather than publishing a `RequestLog` message to the `log` topic the way the primary controllers do. This is deliberate: DLQ controllers must not call back into the primary pipeline. The primary pipeline is what failed and routed the message here in the first place, and re-entering it would risk the same failure mode that caused the DLQ trip. The direct insert produces an equivalent record because the primary log controller is itself a thin wrapper around the same `RequestLogStore`. +When a DLQ controller transitions a request to `RequestStateError`, it publishes the terminal request log to the `log` topic. Gateway consumes that topic and remains the sole writer of the request log and public projections. A publish failure leaves the DLQ delivery unacknowledged so redelivery retries the same idempotent log event. ## Controller mapping @@ -39,8 +39,8 @@ Two controller shapes cover the eleven primary pipeline topics: Reconciliation is safe to run more than once for the same message: -- A request already in a terminal state skips the state-transition CAS but still gets a `RequestStatusError` row appended to the request log. The log write runs unconditionally so that a previous DLQ attempt which successfully flipped the state but then failed to insert the log is repaired on redelivery. A duplicate log entry from such a retry is the accepted trade-off; a missing one would leave the gateway-visible status divergent from the entity state. -- A batch already in a terminal state still fans out to member requests, because a previous attempt may have transitioned the batch but crashed before completing the fan-out. +- A request already in `RequestStateError` skips the state-transition CAS but republishes its terminal log so redelivery repairs an earlier publish failure. Requests in a different terminal state are left unchanged. +- A batch already in `BatchStateFailed` still fans out to member requests, because a previous attempt may have transitioned the batch but crashed before completing the fan-out. Batches in a different terminal state are left unchanged. - Per-request fan-out is itself idempotent via `failRequest`. - A request in `RequestStateCancelling` is reconciled to `RequestStateError`, not left in place: DLQ means the pipeline failed to converge, so we cannot confirm the cancel completed cleanly. Writing Error is the honest signal. diff --git a/submitqueue/orchestrator/controller/dlq/batch.go b/submitqueue/orchestrator/controller/dlq/batch.go index 1d69366a..41309a48 100644 --- a/submitqueue/orchestrator/controller/dlq/batch.go +++ b/submitqueue/orchestrator/controller/dlq/batch.go @@ -40,6 +40,7 @@ type batchController struct { logger *zap.SugaredLogger metricsScope tally.Scope store storage.Storage + registry consumer.TopicRegistry topicKey consumer.TopicKey consumerGroup string } @@ -53,6 +54,7 @@ func NewDLQBatchController( logger *zap.SugaredLogger, scope tally.Scope, store storage.Storage, + registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) consumer.Controller { @@ -61,6 +63,7 @@ func NewDLQBatchController( logger: logger.Named(name), metricsScope: scope.SubScope(name), store: store, + registry: registry, topicKey: topicKey, consumerGroup: consumerGroup, } @@ -94,7 +97,7 @@ func (c *batchController) Process(ctx context.Context, delivery consumer.Deliver "dlq_last_error", dmeta["dlq.last_error"], ) - if err := failBatch(ctx, c.store, c.logger, bid.ID); err != nil { + if err := failBatch(ctx, c.store, c.registry, c.logger, bid.ID, dmeta["dlq.last_error"]); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/batch_test.go b/submitqueue/orchestrator/controller/dlq/batch_test.go index 8fc79edf..826c8e98 100644 --- a/submitqueue/orchestrator/controller/dlq/batch_test.go +++ b/submitqueue/orchestrator/controller/dlq/batch_test.go @@ -32,7 +32,7 @@ func TestDLQBatchController_InterfaceAndAccessors(t *testing.T) { ctrl := gomock.NewController(t) store := storagemock.NewMockStorage(ctrl) - c := NewDLQBatchController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq") + c := NewDLQBatchController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq") assert.Equal(t, "submitqueue-merge_dlq", c.Name()) assert.Equal(t, consumer.TopicKey("submitqueue-merge_dlq"), c.TopicKey()) @@ -55,15 +55,15 @@ func TestDLQBatchController_Process_FailsAndFansOut(t *testing.T) { }, nil) requestStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(1), int32(2), entity.RequestStateError).Return(nil) - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil) + registry := newTestLogRegistry(t, ctrl, 1, func(entity.RequestLog) error { + return nil + }) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - c := NewDLQBatchController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq") + c := NewDLQBatchController(zaptest.NewLogger(t).Sugar(), testScope(), store, registry, TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq") payload, err := entity.BatchID{ID: "q/batch/9"}.ToBytes() require.NoError(t, err) @@ -76,7 +76,7 @@ func TestDLQBatchController_Process_MalformedPayloadFails(t *testing.T) { ctrl := gomock.NewController(t) store := storagemock.NewMockStorage(ctrl) - c := NewDLQBatchController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq") + c := NewDLQBatchController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq") delivery := newMockDelivery(ctrl, []byte("garbage")) err := c.Process(context.Background(), delivery) @@ -87,7 +87,7 @@ func TestDLQBatchController_Process_EmptyIDFails(t *testing.T) { ctrl := gomock.NewController(t) store := storagemock.NewMockStorage(ctrl) - c := NewDLQBatchController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq") + c := NewDLQBatchController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq") payload, err := entity.BatchID{ID: ""}.ToBytes() require.NoError(t, err) diff --git a/submitqueue/orchestrator/controller/dlq/buildsignal.go b/submitqueue/orchestrator/controller/dlq/buildsignal.go index 64ec4600..58a6c5f2 100644 --- a/submitqueue/orchestrator/controller/dlq/buildsignal.go +++ b/submitqueue/orchestrator/controller/dlq/buildsignal.go @@ -39,6 +39,7 @@ type buildSignalController struct { logger *zap.SugaredLogger metricsScope tally.Scope store storage.Storage + registry consumer.TopicRegistry topicKey consumer.TopicKey consumerGroup string } @@ -51,6 +52,7 @@ func NewDLQBuildSignalController( logger *zap.SugaredLogger, scope tally.Scope, store storage.Storage, + registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) consumer.Controller { @@ -59,6 +61,7 @@ func NewDLQBuildSignalController( logger: logger.Named(name), metricsScope: scope.SubScope(name), store: store, + registry: registry, topicKey: topicKey, consumerGroup: consumerGroup, } @@ -118,7 +121,7 @@ func (c *buildSignalController) Process(ctx context.Context, delivery consumer.D return nil } - if err := failBatch(ctx, c.store, c.logger, build.BatchID); err != nil { + if err := failBatch(ctx, c.store, c.registry, c.logger, build.BatchID, dmeta["dlq.last_error"]); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/buildsignal_test.go b/submitqueue/orchestrator/controller/dlq/buildsignal_test.go index 603e58ea..41da60a2 100644 --- a/submitqueue/orchestrator/controller/dlq/buildsignal_test.go +++ b/submitqueue/orchestrator/controller/dlq/buildsignal_test.go @@ -33,7 +33,7 @@ func TestDLQBuildSignalController_InterfaceAndAccessors(t *testing.T) { ctrl := gomock.NewController(t) store := storagemock.NewMockStorage(ctrl) - c := NewDLQBuildSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq") + c := NewDLQBuildSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq") assert.Equal(t, "buildsignal_dlq", c.Name()) assert.Equal(t, consumer.TopicKey("buildsignal_dlq"), c.TopicKey()) @@ -61,16 +61,16 @@ func TestDLQBuildSignalController_Process_FansOutToBatch(t *testing.T) { }, nil) requestStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(1), int32(2), entity.RequestStateError).Return(nil) - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil) + registry := newTestLogRegistry(t, ctrl, 1, func(entity.RequestLog) error { + return nil + }) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - c := NewDLQBuildSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq") + c := NewDLQBuildSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, registry, TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq") payload, err := entity.BuildID{ID: "build-1"}.ToBytes() require.NoError(t, err) @@ -88,7 +88,7 @@ func TestDLQBuildSignalController_Process_BuildNotFoundIsNoOp(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() - c := NewDLQBuildSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq") + c := NewDLQBuildSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq") payload, err := entity.BuildID{ID: "build-1"}.ToBytes() require.NoError(t, err) @@ -108,7 +108,7 @@ func TestDLQBuildSignalController_Process_BuildMissingBatchIsNoOp(t *testing.T) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBuildStore().Return(buildStore).AnyTimes() - c := NewDLQBuildSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq") + c := NewDLQBuildSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq") payload, err := entity.BuildID{ID: "build-1"}.ToBytes() require.NoError(t, err) @@ -121,7 +121,7 @@ func TestDLQBuildSignalController_Process_MalformedPayloadFails(t *testing.T) { ctrl := gomock.NewController(t) store := storagemock.NewMockStorage(ctrl) - c := NewDLQBuildSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq") + c := NewDLQBuildSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq") delivery := newMockDelivery(ctrl, []byte("garbage")) err := c.Process(context.Background(), delivery) diff --git a/submitqueue/orchestrator/controller/dlq/dlq.go b/submitqueue/orchestrator/controller/dlq/dlq.go index f284696e..5de49810 100644 --- a/submitqueue/orchestrator/controller/dlq/dlq.go +++ b/submitqueue/orchestrator/controller/dlq/dlq.go @@ -39,6 +39,7 @@ import ( "fmt" "github.com/uber/submitqueue/platform/consumer" + 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" @@ -60,18 +61,18 @@ func TopicKey(main consumer.TopicKey) consumer.TopicKey { return consumer.TopicKey(string(main) + topicSuffix) } -// failRequest transitions a request to RequestStateError if it is not already -// in a terminal state, and unconditionally appends a RequestStatusError row to -// the request log. The state transition is idempotent — a request already in a -// terminal state skips the UpdateState CAS — but the log insert runs on every -// successful call so that a prior attempt that wrote the Error state but then -// failed to insert the log is repaired on redelivery. +// failRequest transitions a non-terminal request to RequestStateError and +// appends the matching RequestStatusError log. Redelivery for an existing Error +// state repeats materialization to repair a previous partial attempt. A +// different terminal outcome is left unchanged. +// lastError is the failure reason preserved by the queue in DLQ delivery +// metadata and is exposed through Status and History for diagnosis. // // A request in RequestStateCancelling is reconciled to RequestStateError, not // left in place: DLQ means the pipeline failed to converge, so we cannot // confirm the cancel completed cleanly. Writing Error is the honest signal and // keeps the request from being stuck in a non-terminal state forever. -func failRequest(ctx context.Context, store storage.Storage, logger *zap.SugaredLogger, requestID string) error { +func failRequest(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, requestID, lastError string) error { request, err := store.GetRequestStore().Get(ctx, requestID) if err != nil { if errors.Is(err, storage.ErrNotFound) { @@ -84,12 +85,18 @@ func failRequest(ctx context.Context, store storage.Storage, logger *zap.Sugared } logVersion := request.Version - if entity.IsRequestStateTerminal(request.State) { - logger.Infow("dlq reconcile: request already terminal, ensuring log entry", + switch request.State { + case entity.RequestStateError: + logger.Infow("dlq reconcile: request already failed, republishing terminal log", + "request_id", requestID, + ) + case entity.RequestStateLanded, entity.RequestStateCancelled: + logger.Infow("dlq reconcile: request has a different terminal outcome, skipping", "request_id", requestID, "state", string(request.State), ) - } else { + return nil + default: newVersion := request.Version + 1 if err := store.GetRequestStore().UpdateState(ctx, requestID, request.Version, newVersion, entity.RequestStateError); err != nil { return fmt.Errorf("failed to update request %s state to error: %w", requestID, err) @@ -101,24 +108,13 @@ func failRequest(ctx context.Context, store storage.Storage, logger *zap.Sugared ) } - // Append the terminal Error status to the request log directly via the - // RequestLogStore, bypassing the log topic and the primary log controller. - // DLQ controllers must not call back into the primary pipeline (publishing - // to a primary topic) — the primary pipeline is what failed and routed this - // message to the DLQ in the first place, and re-entering it would risk the - // same failure mode that put us here. The log store is the same storage - // backend the primary log controller eventually writes to, so a direct - // insert produces an equivalent record without the round-trip. - // - // The log entry is written unconditionally — even when the state was already - // terminal on entry — so a previous DLQ attempt that succeeded in flipping - // the state but then failed to insert the log is repaired on redelivery. - // A duplicate log entry from such a retry is the accepted trade-off; a - // missing one would leave the gateway-visible status divergent from the - // entity state. - logEntry := entity.NewRequestLog(requestID, entity.RequestStatusError, logVersion, "", nil) - if err := store.GetRequestLogStore().Insert(ctx, logEntry); err != nil { - return fmt.Errorf("failed to insert request log for %s: %w", requestID, err) + // Publish the terminal Error status through the log topic so Gateway remains + // the sole writer of request logs and public projections. An existing Error + // state republishes the same logical event so a previous attempt that changed + // the entity but failed to publish can be repaired. + logEntry := entity.NewRequestLog(requestID, entity.RequestStatusError, logVersion, lastError, nil) + if err := requestcore.PublishLog(ctx, registry, logEntry, requestID); err != nil { + return fmt.Errorf("failed to publish request log for %s: %w", requestID, err) } return nil @@ -135,11 +131,11 @@ func failRequest(ctx context.Context, store storage.Storage, logger *zap.Sugared // same reason failRequest reconciles Cancelling requests: DLQ means we cannot // confirm the cancel completed, so the batch must reach a terminal state. // -// Idempotency: if the batch is already terminal the function still fans out -// to the member requests, because a previous attempt may have transitioned -// the batch but crashed before completing the fan-out. Per-request fan-out -// is itself idempotent via failRequest. -func failBatch(ctx context.Context, store storage.Storage, logger *zap.SugaredLogger, batchID string) error { +// Idempotency: an existing Failed batch repeats fan-out because a previous +// attempt may have crashed after updating the batch. Succeeded and Cancelled +// are different terminal outcomes and do not fan out errors. +// lastError is propagated to each member request's terminal Error log. +func failBatch(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, batchID, lastError string) error { batch, err := store.GetBatchStore().Get(ctx, batchID) if err != nil { if errors.Is(err, storage.ErrNotFound) { @@ -151,12 +147,18 @@ func failBatch(ctx context.Context, store storage.Storage, logger *zap.SugaredLo return fmt.Errorf("failed to get batch %s: %w", batchID, err) } - if batch.State.IsTerminal() { - logger.Infow("dlq reconcile: batch already terminal, fanning out only", + switch batch.State { + case entity.BatchStateFailed: + logger.Infow("dlq reconcile: batch already failed, repairing request fan-out", + "batch_id", batchID, + ) + case entity.BatchStateSucceeded, entity.BatchStateCancelled: + logger.Infow("dlq reconcile: batch has a different terminal outcome, skipping", "batch_id", batchID, "state", string(batch.State), ) - } else { + return nil + default: newVersion := batch.Version + 1 if err := store.GetBatchStore().UpdateState(ctx, batchID, batch.Version, newVersion, entity.BatchStateFailed); err != nil { return fmt.Errorf("failed to update batch %s state to failed: %w", batchID, err) @@ -168,7 +170,7 @@ func failBatch(ctx context.Context, store storage.Storage, logger *zap.SugaredLo } for _, requestID := range batch.Contains { - if err := failRequest(ctx, store, logger, requestID); err != nil { + if err := failRequest(ctx, store, registry, logger, requestID, lastError); err != nil { return fmt.Errorf("fan-out for batch %s: %w", batchID, err) } } diff --git a/submitqueue/orchestrator/controller/dlq/dlq_test.go b/submitqueue/orchestrator/controller/dlq/dlq_test.go index 77600c50..ad8f6ff3 100644 --- a/submitqueue/orchestrator/controller/dlq/dlq_test.go +++ b/submitqueue/orchestrator/controller/dlq/dlq_test.go @@ -22,6 +22,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" @@ -32,43 +33,38 @@ import ( // failRequest -func TestFailRequest_TerminalSkipsUpdateButLogsAnyway(t *testing.T) { - // Truly-terminal states skip the UpdateState CAS but still get a log entry - // written. The log write runs unconditionally so that a previous attempt - // which flipped the state but then failed to insert the log is repaired on - // redelivery. Cancelling is intentionally NOT in this list — see - // TestFailRequest_CancellingTransitionsToError. DLQ must drive every - // non-terminal state to a terminal one or the request stays "in progress" - // forever from the gateway's perspective. - terminalStates := []entity.RequestState{ - entity.RequestStateLanded, - entity.RequestStateError, - entity.RequestStateCancelled, +func TestFailRequest_TerminalStates(t *testing.T) { + tests := []struct { + state entity.RequestState + wantLog bool + }{ + {state: entity.RequestStateLanded}, + {state: entity.RequestStateError, wantLog: true}, + {state: entity.RequestStateCancelled}, } - for _, state := range terminalStates { - state := state - t.Run(string(state), func(t *testing.T) { + + for _, tt := range tests { + t.Run(string(tt.state), func(t *testing.T) { ctrl := gomock.NewController(t) requestStore := storagemock.NewMockRequestStore(ctrl) requestStore.EXPECT().Get(gomock.Any(), "q/1").Return(entity.Request{ - ID: "q/1", Version: 5, State: state, + ID: "q/1", Version: 5, State: tt.state, }, nil) - // no UpdateState expected — state is already terminal - - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.AssignableToTypeOf(entity.RequestLog{})).DoAndReturn(func(_ context.Context, l entity.RequestLog) error { - assert.Equal(t, "q/1", l.RequestID) - assert.Equal(t, entity.RequestStatusError, l.Status) - assert.Equal(t, int32(5), l.RequestVersion) - return nil - }) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - - err := failRequest(context.Background(), store, zaptest.NewLogger(t).Sugar(), "q/1") + registry := consumer.TopicRegistry{} + if tt.wantLog { + registry = newTestLogRegistry(t, ctrl, 1, func(l entity.RequestLog) error { + assert.Equal(t, "q/1", l.RequestID) + assert.Equal(t, entity.RequestStatusError, l.Status) + assert.Equal(t, int32(5), l.RequestVersion) + return nil + }) + } + + err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "") require.NoError(t, err) }) } @@ -88,8 +84,7 @@ func TestFailRequest_CancellingTransitionsToError(t *testing.T) { }, nil) requestStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(7), int32(8), entity.RequestStateError).Return(nil) - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.AssignableToTypeOf(entity.RequestLog{})).DoAndReturn(func(_ context.Context, l entity.RequestLog) error { + registry := newTestLogRegistry(t, ctrl, 1, func(l entity.RequestLog) error { assert.Equal(t, "q/1", l.RequestID) assert.Equal(t, entity.RequestStatusError, l.Status) assert.Equal(t, int32(8), l.RequestVersion) @@ -98,9 +93,8 @@ func TestFailRequest_CancellingTransitionsToError(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - err := failRequest(context.Background(), store, zaptest.NewLogger(t).Sugar(), "q/1") + err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "") require.NoError(t, err) } @@ -113,8 +107,7 @@ func TestFailRequest_TransitionsToError(t *testing.T) { }, nil) requestStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(3), int32(4), entity.RequestStateError).Return(nil) - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.AssignableToTypeOf(entity.RequestLog{})).DoAndReturn(func(_ context.Context, l entity.RequestLog) error { + registry := newTestLogRegistry(t, ctrl, 1, func(l entity.RequestLog) error { assert.Equal(t, "q/1", l.RequestID) assert.Equal(t, entity.RequestStatusError, l.Status) assert.Equal(t, int32(4), l.RequestVersion) @@ -123,18 +116,15 @@ func TestFailRequest_TransitionsToError(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - err := failRequest(context.Background(), store, zaptest.NewLogger(t).Sugar(), "q/1") + err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "") require.NoError(t, err) } -// TestFailRequest_LogInsertErrorPropagates verifies that a failure to append -// the terminal Error row to the request log is surfaced as an error so the -// always-retryable processor will redeliver the DLQ message. We cannot ack a -// half-reconciled request — the state has flipped to Error but the gateway's -// log-store-derived status would still report the prior non-terminal status. -func TestFailRequest_LogInsertErrorPropagates(t *testing.T) { +// TestFailRequest_LogPublishErrorPropagates verifies that a terminal log +// publish failure is surfaced so the always-retryable processor redelivers the +// DLQ message. +func TestFailRequest_LogPublishErrorPropagates(t *testing.T) { ctrl := gomock.NewController(t) requestStore := storagemock.NewMockRequestStore(ctrl) @@ -143,14 +133,14 @@ func TestFailRequest_LogInsertErrorPropagates(t *testing.T) { }, nil) requestStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(3), int32(4), entity.RequestStateError).Return(nil) - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(fmt.Errorf("log store boom")) + registry := newTestLogRegistry(t, ctrl, 1, func(entity.RequestLog) error { + return fmt.Errorf("publish boom") + }) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - err := failRequest(context.Background(), store, zaptest.NewLogger(t).Sugar(), "q/1") + err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "") require.Error(t, err) } @@ -163,7 +153,7 @@ func TestFailRequest_NotFoundIsNoOp(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failRequest(context.Background(), store, zaptest.NewLogger(t).Sugar(), "q/1") + err := failRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", "") require.NoError(t, err) } @@ -176,7 +166,7 @@ func TestFailRequest_GenericGetErrorIsNonRetryable(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - err := failRequest(context.Background(), store, zaptest.NewLogger(t).Sugar(), "q/1") + err := failRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", "") require.Error(t, err) assert.False(t, errs.IsRetryable(err)) } @@ -203,22 +193,19 @@ func TestFailBatch_TransitionsAndFansOut(t *testing.T) { }, nil) requestStore.EXPECT().UpdateState(gomock.Any(), "q/2", int32(1), int32(2), entity.RequestStateError).Return(nil) - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil).Times(2) + registry := newTestLogRegistry(t, ctrl, 2, func(entity.RequestLog) error { + return nil + }) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - err := failBatch(context.Background(), store, zaptest.NewLogger(t).Sugar(), "q/batch/1") + err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") require.NoError(t, err) } -func TestFailBatch_AlreadyTerminalFansOutOnly(t *testing.T) { - // Already-terminal batch: skip the CAS but still drive the fan-out so a - // prior crashed attempt that updated the batch but not the requests still - // reconciles to a clean terminal state. +func TestFailBatch_FailedFansOutForRepair(t *testing.T) { ctrl := gomock.NewController(t) batchStore := storagemock.NewMockBatchStore(ctrl) @@ -234,18 +221,36 @@ func TestFailBatch_AlreadyTerminalFansOutOnly(t *testing.T) { }, nil) requestStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(2), int32(3), entity.RequestStateError).Return(nil) - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil) + registry := newTestLogRegistry(t, ctrl, 1, func(entity.RequestLog) error { + return nil + }) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - err := failBatch(context.Background(), store, zaptest.NewLogger(t).Sugar(), "q/batch/1") + err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") require.NoError(t, err) } +func TestFailBatch_DifferentTerminalOutcomeSkipsFanOut(t *testing.T) { + for _, state := range []entity.BatchState{entity.BatchStateSucceeded, entity.BatchStateCancelled} { + t.Run(string(state), func(t *testing.T) { + ctrl := gomock.NewController(t) + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), "q/batch/1").Return(entity.Batch{ + ID: "q/batch/1", Queue: "q", Contains: []string{"q/1"}, State: state, Version: 5, + }, nil) + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + + err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") + require.NoError(t, err) + }) + } +} + // TestFailBatch_CancellingTransitionsToFailed verifies that a batch stuck in // the non-terminal Cancelling state is reconciled to Failed and its member // requests are driven from Cancelling to Error. Same rationale as @@ -268,15 +273,15 @@ func TestFailBatch_CancellingTransitionsToFailed(t *testing.T) { }, nil) requestStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(3), int32(4), entity.RequestStateError).Return(nil) - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil) + registry := newTestLogRegistry(t, ctrl, 1, func(entity.RequestLog) error { + return nil + }) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - err := failBatch(context.Background(), store, zaptest.NewLogger(t).Sugar(), "q/batch/1") + err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") require.NoError(t, err) } @@ -289,7 +294,7 @@ func TestFailBatch_NotFoundIsNoOp(t *testing.T) { store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - err := failBatch(context.Background(), store, zaptest.NewLogger(t).Sugar(), "q/batch/1") + err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", "") require.NoError(t, err) } diff --git a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go index 8bb18695..26f0fa7f 100644 --- a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go +++ b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go @@ -34,6 +34,7 @@ type mergeConflictSignalController struct { logger *zap.SugaredLogger metricsScope tally.Scope store storage.Storage + registry consumer.TopicRegistry topicKey consumer.TopicKey consumerGroup string } @@ -47,6 +48,7 @@ func NewDLQMergeConflictSignalController( logger *zap.SugaredLogger, scope tally.Scope, store storage.Storage, + registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) consumer.Controller { @@ -55,6 +57,7 @@ func NewDLQMergeConflictSignalController( logger: logger.Named(name), metricsScope: scope.SubScope(name), store: store, + registry: registry, topicKey: topicKey, consumerGroup: consumerGroup, } @@ -84,7 +87,7 @@ func (c *mergeConflictSignalController) Process(ctx context.Context, delivery co "dlq_last_error", dmeta["dlq.last_error"], ) - if err := failRequest(ctx, c.store, c.logger, result.Id); err != nil { + if err := failRequest(ctx, c.store, c.registry, c.logger, result.Id, dmeta["dlq.last_error"]); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal_test.go b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal_test.go index bf03e459..87a35514 100644 --- a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal_test.go +++ b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal_test.go @@ -33,7 +33,7 @@ func TestDLQMergeConflictSignalController_InterfaceAndAccessors(t *testing.T) { ctrl := gomock.NewController(t) store := storagemock.NewMockStorage(ctrl) - c := NewDLQMergeConflictSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(runwaymq.TopicKeyMergeConflictCheckSignal), "orchestrator-mergeconflictsignal-dlq") + c := NewDLQMergeConflictSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, TopicKey(runwaymq.TopicKeyMergeConflictCheckSignal), "orchestrator-mergeconflictsignal-dlq") assert.Equal(t, "merge-conflict-check-signal_dlq", c.Name()) assert.Equal(t, consumer.TopicKey("merge-conflict-check-signal_dlq"), c.TopicKey()) @@ -49,14 +49,14 @@ func TestDLQMergeConflictSignalController_Process_ReconcilesRequest(t *testing.T }, nil) requestStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(1), int32(2), entity.RequestStateError).Return(nil) - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil) + registry := newTestLogRegistry(t, ctrl, 1, func(entity.RequestLog) error { + return nil + }) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - c := NewDLQMergeConflictSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(runwaymq.TopicKeyMergeConflictCheckSignal), "orchestrator-mergeconflictsignal-dlq") + c := NewDLQMergeConflictSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, registry, TopicKey(runwaymq.TopicKeyMergeConflictCheckSignal), "orchestrator-mergeconflictsignal-dlq") payload, err := runwaymq.Marshal(&runwaymq.MergeResult{Id: "q/1", Outcome: runwaypb.Outcome_FAILED, Reason: "boom"}) require.NoError(t, err) @@ -69,7 +69,7 @@ func TestDLQMergeConflictSignalController_Process_MalformedPayloadFails(t *testi ctrl := gomock.NewController(t) store := storagemock.NewMockStorage(ctrl) - c := NewDLQMergeConflictSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(runwaymq.TopicKeyMergeConflictCheckSignal), "orchestrator-mergeconflictsignal-dlq") + c := NewDLQMergeConflictSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, TopicKey(runwaymq.TopicKeyMergeConflictCheckSignal), "orchestrator-mergeconflictsignal-dlq") delivery := newMockDelivery(ctrl, []byte("garbage")) require.Error(t, c.Process(context.Background(), delivery)) diff --git a/submitqueue/orchestrator/controller/dlq/mergesignal.go b/submitqueue/orchestrator/controller/dlq/mergesignal.go index af103c85..fff67ce1 100644 --- a/submitqueue/orchestrator/controller/dlq/mergesignal.go +++ b/submitqueue/orchestrator/controller/dlq/mergesignal.go @@ -34,6 +34,7 @@ type mergeSignalController struct { logger *zap.SugaredLogger metricsScope tally.Scope store storage.Storage + registry consumer.TopicRegistry topicKey consumer.TopicKey consumerGroup string } @@ -46,6 +47,7 @@ func NewDLQMergeSignalController( logger *zap.SugaredLogger, scope tally.Scope, store storage.Storage, + registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) consumer.Controller { @@ -54,6 +56,7 @@ func NewDLQMergeSignalController( logger: logger.Named(name), metricsScope: scope.SubScope(name), store: store, + registry: registry, topicKey: topicKey, consumerGroup: consumerGroup, } @@ -83,7 +86,7 @@ func (c *mergeSignalController) Process(ctx context.Context, delivery consumer.D "dlq_last_error", dmeta["dlq.last_error"], ) - if err := failBatch(ctx, c.store, c.logger, result.Id); err != nil { + if err := failBatch(ctx, c.store, c.registry, c.logger, result.Id, dmeta["dlq.last_error"]); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/mergesignal_test.go b/submitqueue/orchestrator/controller/dlq/mergesignal_test.go index cc105c17..b44257eb 100644 --- a/submitqueue/orchestrator/controller/dlq/mergesignal_test.go +++ b/submitqueue/orchestrator/controller/dlq/mergesignal_test.go @@ -33,7 +33,7 @@ func TestDLQMergeSignalController_InterfaceAndAccessors(t *testing.T) { ctrl := gomock.NewController(t) store := storagemock.NewMockStorage(ctrl) - c := NewDLQMergeSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(runwaymq.TopicKeyMergeSignal), "orchestrator-mergesignal-dlq") + c := NewDLQMergeSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, TopicKey(runwaymq.TopicKeyMergeSignal), "orchestrator-mergesignal-dlq") assert.Equal(t, "merge-signal_dlq", c.Name()) assert.Equal(t, consumer.TopicKey("merge-signal_dlq"), c.TopicKey()) @@ -58,15 +58,15 @@ func TestDLQMergeSignalController_Process_ReconcilesBatch(t *testing.T) { }, nil) requestStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(1), int32(2), entity.RequestStateError).Return(nil) - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil) + registry := newTestLogRegistry(t, ctrl, 1, func(entity.RequestLog) error { + return nil + }) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - c := NewDLQMergeSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(runwaymq.TopicKeyMergeSignal), "orchestrator-mergesignal-dlq") + c := NewDLQMergeSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, registry, TopicKey(runwaymq.TopicKeyMergeSignal), "orchestrator-mergesignal-dlq") payload, err := runwaymq.Marshal(&runwaymq.MergeResult{Id: "q/batch/1", Outcome: runwaypb.Outcome_FAILED, Reason: "boom"}) require.NoError(t, err) @@ -79,7 +79,7 @@ func TestDLQMergeSignalController_Process_MalformedPayloadFails(t *testing.T) { ctrl := gomock.NewController(t) store := storagemock.NewMockStorage(ctrl) - c := NewDLQMergeSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, TopicKey(runwaymq.TopicKeyMergeSignal), "orchestrator-mergesignal-dlq") + c := NewDLQMergeSignalController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, TopicKey(runwaymq.TopicKeyMergeSignal), "orchestrator-mergesignal-dlq") delivery := newMockDelivery(ctrl, []byte("garbage")) require.Error(t, c.Process(context.Background(), delivery)) diff --git a/submitqueue/orchestrator/controller/dlq/publisher_test.go b/submitqueue/orchestrator/controller/dlq/publisher_test.go new file mode 100644 index 00000000..ea4192ff --- /dev/null +++ b/submitqueue/orchestrator/controller/dlq/publisher_test.go @@ -0,0 +1,54 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dlq + +import ( + "context" + + "github.com/stretchr/testify/require" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + "github.com/uber/submitqueue/submitqueue/core/topickey" + "github.com/uber/submitqueue/submitqueue/entity" + "go.uber.org/mock/gomock" +) + +func newTestLogRegistry( + t require.TestingT, + ctrl *gomock.Controller, + publishCount int, + publishFn func(entity.RequestLog) error, +) consumer.TopicRegistry { + publisher := queuemock.NewMockPublisher(ctrl) + publisher.EXPECT().Publish(gomock.Any(), "log", gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, message entityqueue.Message) error { + logEntry, err := entity.RequestLogFromBytes(message.Payload) + require.NoError(t, err) + return publishFn(logEntry) + }, + ).Times(publishCount) + + queue := queuemock.NewMockQueue(ctrl) + queue.EXPECT().Publisher().Return(publisher).Times(publishCount) + + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{{ + Key: topickey.TopicKeyLog, + Name: "log", + Queue: queue, + }}) + require.NoError(t, err) + return registry +} diff --git a/submitqueue/orchestrator/controller/dlq/request.go b/submitqueue/orchestrator/controller/dlq/request.go index a5b728a8..b00244be 100644 --- a/submitqueue/orchestrator/controller/dlq/request.go +++ b/submitqueue/orchestrator/controller/dlq/request.go @@ -72,6 +72,7 @@ type requestController struct { logger *zap.SugaredLogger metricsScope tally.Scope store storage.Storage + registry consumer.TopicRegistry decode RequestIDDecoder topicKey consumer.TopicKey consumerGroup string @@ -87,6 +88,7 @@ func NewDLQRequestController( logger *zap.SugaredLogger, scope tally.Scope, store storage.Storage, + registry consumer.TopicRegistry, decode RequestIDDecoder, topicKey consumer.TopicKey, consumerGroup string, @@ -96,6 +98,7 @@ func NewDLQRequestController( logger: logger.Named(name), metricsScope: scope.SubScope(name), store: store, + registry: registry, decode: decode, topicKey: topicKey, consumerGroup: consumerGroup, @@ -134,7 +137,7 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv "dlq_last_error", dmeta["dlq.last_error"], ) - if err := failRequest(ctx, c.store, c.logger, requestID); err != nil { + if err := failRequest(ctx, c.store, c.registry, c.logger, requestID, dmeta["dlq.last_error"]); err != nil { metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1) return err } diff --git a/submitqueue/orchestrator/controller/dlq/request_test.go b/submitqueue/orchestrator/controller/dlq/request_test.go index 4dcfc1b8..17fc0eab 100644 --- a/submitqueue/orchestrator/controller/dlq/request_test.go +++ b/submitqueue/orchestrator/controller/dlq/request_test.go @@ -34,7 +34,7 @@ func TestDLQRequestController_InterfaceAndAccessors(t *testing.T) { ctrl := gomock.NewController(t) store := storagemock.NewMockStorage(ctrl) - c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, DecodeRequestID, TopicKey(topickey.TopicKeyValidate), "orchestrator-validate-dlq") + c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, DecodeRequestID, TopicKey(topickey.TopicKeyValidate), "orchestrator-validate-dlq") assert.Equal(t, "validate_dlq", c.Name()) assert.Equal(t, consumer.TopicKey("validate_dlq"), c.TopicKey()) @@ -50,14 +50,14 @@ func TestDLQRequestController_Process_LandRequestPayload(t *testing.T) { }, nil) requestStore.EXPECT().UpdateState(gomock.Any(), "q/1", int32(1), int32(2), entity.RequestStateError).Return(nil) - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil) + registry := newTestLogRegistry(t, ctrl, 1, func(entity.RequestLog) error { + return nil + }) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, DecodeLandRequestID, TopicKey(topickey.TopicKeyStart), "orchestrator-start-dlq") + c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, registry, DecodeLandRequestID, TopicKey(topickey.TopicKeyStart), "orchestrator-start-dlq") payload, err := entity.LandRequest{ID: "q/1", Queue: "q"}.ToBytes() require.NoError(t, err) @@ -75,14 +75,14 @@ func TestDLQRequestController_Process_CancelRequestPayload(t *testing.T) { }, nil) requestStore.EXPECT().UpdateState(gomock.Any(), "q/7", int32(2), int32(3), entity.RequestStateError).Return(nil) - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil) + registry := newTestLogRegistry(t, ctrl, 1, func(entity.RequestLog) error { + return nil + }) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, DecodeCancelRequestID, TopicKey(topickey.TopicKeyCancel), "orchestrator-cancel-dlq") + c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, registry, DecodeCancelRequestID, TopicKey(topickey.TopicKeyCancel), "orchestrator-cancel-dlq") payload, err := entity.CancelRequest{ID: "q/7", Reason: "user"}.ToBytes() require.NoError(t, err) @@ -100,14 +100,15 @@ func TestDLQRequestController_Process_RequestIDPayload(t *testing.T) { }, nil) requestStore.EXPECT().UpdateState(gomock.Any(), "q/3", int32(1), int32(2), entity.RequestStateError).Return(nil) - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil) + registry := newTestLogRegistry(t, ctrl, 1, func(log entity.RequestLog) error { + assert.Equal(t, "boom", log.LastError) + return nil + }) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, DecodeRequestID, TopicKey(topickey.TopicKeyBatch), "orchestrator-batch-dlq") + c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, registry, DecodeRequestID, TopicKey(topickey.TopicKeyBatch), "orchestrator-batch-dlq") payload, err := entity.RequestID{ID: "q/3"}.ToBytes() require.NoError(t, err) @@ -116,23 +117,18 @@ func TestDLQRequestController_Process_RequestIDPayload(t *testing.T) { require.NoError(t, c.Process(context.Background(), delivery)) } -func TestDLQRequestController_Process_AlreadyTerminalSkipsUpdateButLogsAnyway(t *testing.T) { +func TestDLQRequestController_Process_DifferentTerminalOutcomeSkips(t *testing.T) { ctrl := gomock.NewController(t) requestStore := storagemock.NewMockRequestStore(ctrl) requestStore.EXPECT().Get(gomock.Any(), "q/1").Return(entity.Request{ ID: "q/1", Version: 5, State: entity.RequestStateLanded, }, nil) - // no UpdateState expected - - logStore := storagemock.NewMockRequestLogStore(ctrl) - logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() - store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, DecodeRequestID, TopicKey(topickey.TopicKeyValidate), "orchestrator-validate-dlq") + c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, DecodeRequestID, TopicKey(topickey.TopicKeyValidate), "orchestrator-validate-dlq") payload, err := entity.RequestID{ID: "q/1"}.ToBytes() require.NoError(t, err) @@ -147,7 +143,7 @@ func TestDLQRequestController_Process_MalformedPayloadFails(t *testing.T) { store := storagemock.NewMockStorage(ctrl) // no store calls expected - c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, DecodeRequestID, TopicKey(topickey.TopicKeyValidate), "orchestrator-validate-dlq") + c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, DecodeRequestID, TopicKey(topickey.TopicKeyValidate), "orchestrator-validate-dlq") delivery := newMockDelivery(ctrl, []byte("not json")) err := c.Process(context.Background(), delivery) @@ -160,7 +156,7 @@ func TestDLQRequestController_Process_EmptyIDFails(t *testing.T) { store := storagemock.NewMockStorage(ctrl) // no store calls expected - c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, DecodeRequestID, TopicKey(topickey.TopicKeyValidate), "orchestrator-validate-dlq") + c := NewDLQRequestController(zaptest.NewLogger(t).Sugar(), testScope(), store, consumer.TopicRegistry{}, DecodeRequestID, TopicKey(topickey.TopicKeyValidate), "orchestrator-validate-dlq") payload, err := entity.RequestID{ID: ""}.ToBytes() require.NoError(t, err) diff --git a/test/integration/submitqueue/gateway/BUILD.bazel b/test/integration/submitqueue/gateway/BUILD.bazel index cdc8b010..36fda2d6 100644 --- a/test/integration/submitqueue/gateway/BUILD.bazel +++ b/test/integration/submitqueue/gateway/BUILD.bazel @@ -24,6 +24,7 @@ go_test( "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", + "//submitqueue/extension/storage/mysql:go_default_library", "//test/testutil:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", diff --git a/test/integration/submitqueue/gateway/suite_test.go b/test/integration/submitqueue/gateway/suite_test.go index f14dd03a..1c0c3899 100644 --- a/test/integration/submitqueue/gateway/suite_test.go +++ b/test/integration/submitqueue/gateway/suite_test.go @@ -44,6 +44,7 @@ import ( corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" + mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql" "github.com/uber/submitqueue/test/testutil" "go.uber.org/zap" "google.golang.org/grpc" @@ -192,6 +193,13 @@ func (s *GatewayIntegrationSuite) TestRequestLogConsumer() { require.NoError(t, err, "failed to create topic registry") const sqid = "log-consumer-test/1" + store, err := mysqlstorage.NewStorage(s.db, tally.NoopScope) + require.NoError(t, err) + summary := entity.RequestSummary{ + RequestID: sqid, Queue: "log-consumer-test", ChangeURIs: []string{}, ReceivedAtMs: 1, + Status: entity.RequestStatusAccepting, StatusTimestampMs: 1, Version: 1, Metadata: map[string]string{}, + } + require.NoError(t, store.GetRequestSummaryStore().Create(s.ctx, summary)) logEntry := entity.NewRequestLog(sqid, entity.RequestStatusStarted, 1, "", nil) require.NoError(t, corerequest.PublishLog(s.ctx, registry, logEntry, sqid), "failed to publish request log to log topic")