diff --git a/doc/rfc/submitqueue/workflow.md b/doc/rfc/submitqueue/workflow.md index 2bd408937..53cff31a7 100644 --- a/doc/rfc/submitqueue/workflow.md +++ b/doc/rfc/submitqueue/workflow.md @@ -158,7 +158,7 @@ class node_submitter toneIndigo ## DLQ reconciliation -Every consumed stage in `pipeline.go` is paired with a `{topic}_dlq` subscription. `pipeline.Construct` registers that companion with `errs.AlwaysRetryableProcessor` and `DLQSubscriptionConfig`, which disables a second-level DLQ and sets `Retry.MaxAttempts` to 0 (unlimited). The consumer moves a message to its DLQ once the primary controller returns a non-retryable error or exhausts retries on a retryable one. The publish-only topics — `log`, Runway `merge-conflict-check`, and Runway `runway-merge` — have no orchestrator subscription and therefore no orchestrator DLQ. The gateway consumes `log`. Runway consumes the two merge request topics. +Every consumed stage in `pipeline.go` is paired with a `{topic}_dlq` subscription. `pipeline.Construct` registers that companion with `errs.AlwaysRetryableProcessor` and `DLQSubscriptionConfig`, which disables a second-level DLQ and sets `Retry.MaxAttempts` to 0 (unlimited). The consumer moves a message to its DLQ once the primary controller returns a non-retryable error or exhausts retries on a retryable one. The publish-only topics — `log`, Runway `merge-conflict-check`, and Runway `runway-merge` — have no orchestrator subscription and therefore no orchestrator DLQ. The gateway consumes `log` and owns its `log_dlq` recovery subscription. Runway consumes the two merge request topics. Most DLQ controllers do not re-attempt the failed work. They decode the payload to a `RequestID` or `BatchID` and drive that entity to a terminal failed state — `RequestStateError` for requests, `BatchStateFailed` for batches, with fan-out to the member requests. Topics that carry a full payload recover the id from it: the `landconflictsignal` DLQ reads the request id from Runway's `MergeResult`, and the `landsignal` DLQ reads the batch id. The `buildsignal` DLQ decodes a build id, loads that build, and fails the batch that owns it. A missing build or an empty batch id is acked: there is no batch to fail from this signal. State writes use the same optimistic-locking CAS as the primary pipeline, so a late primary-pipeline update wins cleanly and a version mismatch is asked back for redelivery. @@ -203,3 +203,7 @@ Rejected alternatives: - **UUIDs or one global counter.** They buy global uniqueness at the cost of readable IDs or cross-queue coordination. - **Integer wire fields.** They would tie the API to the counter's representation without adding meaning. - **SQL auto-increment, `MAX(id) + 1`, or process-local counters.** These put allocation inside one store, race under concurrency, or reuse IDs after a restart. + +The gateway replays `log_dlq` through its primary log controller and materializer, rather than asking the orchestrator to write gateway projections. Valid log recovery acknowledges only after all public read-model writes succeed; unlimited retries and capped backoff keep failed writes durable. With valid envelope metadata, malformed payloads and payload-queue conflicts remain in the final queue, with structured diagnostics and counters; retrying unchanged bad data does not repair it. Retention is not isolated quarantine: poison entries can impede later work. Before deleting a bad record, operators must durably export it and record a disposition; corrected records re-enter gateway materialization. Replays preserve equivalent audit-history rows while summary precedence and incomplete-mapping repair remain unchanged. Rollback leaves final-DLQ rows intact. + +Invalid envelope metadata is instead logged and rejected before controller processing. It is a permanent error while unchanged; without another DLQ, rejection may acknowledge and discard the delivery rather than retry it. diff --git a/service/submitqueue/gateway/server/BUILD.bazel b/service/submitqueue/gateway/server/BUILD.bazel index 0a2d9c205..5e1709930 100644 --- a/service/submitqueue/gateway/server/BUILD.bazel +++ b/service/submitqueue/gateway/server/BUILD.bazel @@ -12,7 +12,10 @@ exports_files( go_library( name = "gateway_lib", - srcs = ["main.go"], + srcs = [ + "log_consumer.go", + "main.go", + ], importpath = "github.com/uber/submitqueue/service/submitqueue/gateway/server", visibility = ["//visibility:private"], deps = [ @@ -28,6 +31,7 @@ go_library( "//platform/extension/counter/mysql:go_default_library", "//platform/extension/messagequeue:go_default_library", "//platform/extension/messagequeue/mysql:go_default_library", + "//platform/lifecycle:go_default_library", "//service/messagequeue:go_default_library", "//service/submitqueue/core:go_default_library", "//service/submitqueue/gateway/server/mapper:go_default_library", @@ -83,17 +87,32 @@ filegroup( go_test( name = "go_default_test", - srcs = ["main_test.go"], + srcs = [ + "log_consumer_test.go", + "main_test.go", + ], embed = [":gateway_lib"], # keep deps = [ "//api/submitqueue/gateway/protopb:go_default_library", + "//platform/base/failure:go_default_library", + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", "//platform/errs:go_default_library", + "//platform/extension/consumergate/noop:go_default_library", + "//platform/extension/messagequeue:go_default_library", + "//platform/extension/messagequeue/mock:go_default_library", + "//platform/lifecycle:go_default_library", + "//submitqueue/core/messagequeue:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/queueconfig/mock:go_default_library", "//submitqueue/gateway/controller:go_default_library", + "//submitqueue/gateway/core/request:go_default_library", + "//submitqueue/gateway/extension/storage:go_default_library", + "//submitqueue/gateway/extension/storage/mock:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", "@com_github_uber_go_tally//:go_default_library", + "@org_golang_google_grpc//:go_default_library", "@org_golang_google_grpc//codes:go_default_library", "@org_golang_google_grpc//status:go_default_library", "@org_uber_go_mock//gomock:go_default_library", diff --git a/service/submitqueue/gateway/server/log_consumer.go b/service/submitqueue/gateway/server/log_consumer.go new file mode 100644 index 000000000..b52e23a68 --- /dev/null +++ b/service/submitqueue/gateway/server/log_consumer.go @@ -0,0 +1,91 @@ +// 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 main + +import ( + "context" + "fmt" + "time" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/errs" + genericerrs "github.com/uber/submitqueue/platform/errs/generic" + mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql" + "github.com/uber/submitqueue/platform/extension/consumergate" + extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" + "github.com/uber/submitqueue/platform/lifecycle" + "github.com/uber/submitqueue/submitqueue/core/topickey" + logctrl "github.com/uber/submitqueue/submitqueue/gateway/controller/log" + requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request" + "go.uber.org/zap" +) + +const ( + logDLQSuffix = "_dlq" + logDLQTopicKey = topickey.TopicKeyLog + consumer.TopicKey(logDLQSuffix) + logConsumerGroup = "gateway-log" + logDLQConsumerGroup = "gateway-log-dlq" +) + +func newGatewayTopicRegistry(queue extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) { + primary := extqueue.DefaultSubscriptionConfig(subscriberName, logConsumerGroup) + primary.DLQ.TopicSuffix = logDLQSuffix + return consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: topickey.TopicKeyStart, Name: "start", Queue: queue}, + {Key: topickey.TopicKeyCancel, Name: "cancel", Queue: queue}, + {Key: topickey.TopicKeyLog, Name: "log", Queue: queue, Subscription: primary}, + { + Key: logDLQTopicKey, Name: "log" + logDLQSuffix, Queue: queue, + Subscription: extqueue.DLQSubscriptionConfig(subscriberName, logDLQConsumerGroup), + }, + }) +} + +func newLogConsumers(logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, materializer *requestcore.Materializer, gate consumergate.Gate) (*lifecycle.Group, error) { + // Both lanes share the controller/materializer. One always-retry primary with + // unlimited attempts and no DLQ is an alternative. Separate recovery keeps failed + // logs off normal traffic; in the SQL backend, the primary acknowledgment/cleanup + // checkpoint can then advance past them. + policies := []struct { + topic consumer.TopicKey + group string + processor errs.ErrorProcessor + }{ + {topickey.TopicKeyLog, logConsumerGroup, errs.NewClassifierProcessor(genericerrs.Classifier, mysqlerrs.Classifier)}, + {logDLQTopicKey, logDLQConsumerGroup, errs.AlwaysRetryableProcessor}, + } + components := make([]lifecycle.Component, 0, len(policies)) + for _, policy := range policies { + c := consumer.New(logger, scope.SubScope("consumer"), registry, policy.processor, gate) + if err := c.Register(logctrl.NewController(logger, scope, materializer, policy.topic, policy.group)); err != nil { + return nil, fmt.Errorf("failed to register log controller: %w", err) + } + components = append(components, logConsumerLifecycle{Consumer: c}) + } + return lifecycle.NewGroup(components...), nil +} + +type logConsumerLifecycle struct { + consumer.Consumer +} + +func (c logConsumerLifecycle) Stop(ctx context.Context) error { + timeout := 30 * time.Second + if deadline, ok := ctx.Deadline(); ok { + timeout = min(timeout, max(time.Until(deadline), 0)) + } + return c.Consumer.Stop(timeout.Milliseconds()) +} diff --git a/service/submitqueue/gateway/server/log_consumer_test.go b/service/submitqueue/gateway/server/log_consumer_test.go new file mode 100644 index 000000000..ad52e957e --- /dev/null +++ b/service/submitqueue/gateway/server/log_consumer_test.go @@ -0,0 +1,175 @@ +// 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 main + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/base/failure" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop" + extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" + queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + "github.com/uber/submitqueue/platform/lifecycle" + sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue" + "github.com/uber/submitqueue/submitqueue/entity" + requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request" + gwstorage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" + gwstoragemock "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" + "google.golang.org/grpc" +) + +func TestLogConsumers_RetainFailedRecovery(t *testing.T) { + for _, tt := range []struct { + name string + topic string + malformed bool + wrongQueue bool + }{ + {name: "primary rejects unclassified failure", topic: "log"}, + {name: "final retries unclassified failure", topic: "log_dlq"}, + {name: "final retains malformed payload", topic: "log_dlq", malformed: true}, + {name: "final retains payload identity mismatch", topic: "log_dlq", wrongQueue: true}, + } { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + primary := make(chan extqueue.Delivery, 1) + final := make(chan extqueue.Delivery, 1) + subscriber := queuemock.NewMockSubscriber(ctrl) + subscriber.EXPECT().Subscribe(gomock.Any(), "log", gomock.Any()).DoAndReturn(func(_ context.Context, _ string, config extqueue.SubscriptionConfig) (<-chan extqueue.Delivery, error) { + assert.Equal(t, "gateway-log", config.ConsumerGroup) + assert.True(t, config.DLQ.Enabled) + assert.Equal(t, "_dlq", config.DLQ.TopicSuffix) + assert.Positive(t, config.Retry.MaxAttempts) + return primary, nil + }) + subscriber.EXPECT().Subscribe(gomock.Any(), "log_dlq", gomock.Any()).DoAndReturn(func(_ context.Context, _ string, config extqueue.SubscriptionConfig) (<-chan extqueue.Delivery, error) { + assert.Equal(t, "gateway-log-dlq", config.ConsumerGroup) + assert.False(t, config.DLQ.Enabled) + assert.Zero(t, config.Retry.MaxAttempts) + assert.Positive(t, config.Retry.MaxBackoffMs) + return final, nil + }) + queue := queuemock.NewMockQueue(ctrl) + queue.EXPECT().Subscriber().Return(subscriber).Times(2) + registry, err := newGatewayTopicRegistry(queue, "worker") + require.NoError(t, err) + factory := gwstoragemock.NewMockFactory(ctrl) + if !tt.malformed && !tt.wrongQueue { + factory.EXPECT().For(gwstorage.Config{QueueName: "queue-a"}).Return(nil, errors.New("storage unavailable")) + } + consumers, err := newLogConsumers(zap.NewNop().Sugar(), tally.NoopScope, registry, requestcore.NewMaterializer(factory), consumergatenoop.New()) + require.NoError(t, err) + require.NoError(t, consumers.Start(context.Background())) + t.Cleanup(func() { require.NoError(t, consumers.Stop(context.Background())) }) + + log := entity.NewRequestStatusLog("queue-a", "1", entity.RequestStatusLanded, 3, "", nil) + payload, err := sqmq.Marshal(sqmq.LogFromEntity(log)) + require.NoError(t, err) + if tt.malformed { + payload = []byte("not JSON") + } + msg := entityqueue.NewMessage("terminal-log", payload, "queue-a", nil) + msg.Tenant = "queue-a" + if tt.wrongQueue { + msg.Tenant = "queue-b" + } + delivery := queuemock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(msg).AnyTimes() + delivery.EXPECT().Attempt().Return(100).AnyTimes() + delivery.EXPECT().Metadata().Return(nil).AnyTimes() + settled := make(chan struct{}) + settle := func(context.Context, failure.Failure) error { close(settled); return nil } + if tt.topic == "log" { + delivery.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(settle) + primary <- delivery + } else { + delivery.EXPECT().Nack(gomock.Any(), gomock.Any()).DoAndReturn(settle) + final <- delivery + } + <-settled + }) + } +} + +type lifecycleTestConsumer struct { + consumer.Consumer + name string + calls *[]string + startErr error + stopErr error +} + +func (c lifecycleTestConsumer) Start(context.Context) error { + *c.calls = append(*c.calls, "start:"+c.name) + return c.startErr +} + +func (c lifecycleTestConsumer) Stop(timeoutMs int64) error { + *c.calls = append(*c.calls, "stop:"+c.name) + return c.stopErr +} + +func TestServeGateway_ConsumerCleanup(t *testing.T) { + startErr := errors.New("subscribe unavailable") + stopErr := errors.New("stop unavailable") + for _, tt := range []struct { + name string + primaryErr error + recoveryErr error + stopErr error + cancelled bool + want []string + }{ + {name: "primary fails", primaryErr: startErr, want: []string{"start:primary"}}, + {name: "recovery fails and rollback fails", recoveryErr: startErr, stopErr: stopErr, want: []string{"start:primary", "start:recovery", "stop:primary"}}, + {name: "listener fails and both stop despite cleanup failure", stopErr: stopErr, want: []string{"start:primary", "start:recovery", "stop:recovery", "stop:primary"}}, + {name: "signal drains both", cancelled: true, want: []string{"start:primary", "start:recovery", "stop:recovery", "stop:primary"}}, + {name: "signal does not mask shutdown failure", cancelled: true, stopErr: stopErr, want: []string{"start:primary", "start:recovery", "stop:recovery", "stop:primary"}}, + } { + t.Run(tt.name, func(t *testing.T) { + var calls []string + group := lifecycle.NewGroup( + logConsumerLifecycle{Consumer: lifecycleTestConsumer{name: "primary", calls: &calls, startErr: tt.primaryErr, stopErr: tt.stopErr}}, + logConsumerLifecycle{Consumer: lifecycleTestConsumer{name: "recovery", calls: &calls, startErr: tt.recoveryErr, stopErr: tt.stopErr}}, + ) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + t.Setenv("PORT", "invalid-port") + if tt.cancelled { + t.Setenv("PORT", "127.0.0.1:0") + cancel() + } + err := serveGateway(ctx, cancel, grpc.NewServer(), group) + require.Error(t, err) + if tt.primaryErr != nil || tt.recoveryErr != nil { + assert.ErrorIs(t, err, startErr) + } + if tt.stopErr != nil { + assert.ErrorIs(t, err, stopErr) + } + assert.Equal(t, tt.cancelled && tt.stopErr == nil, errors.Is(err, context.Canceled)) + assert.Equal(t, tt.want, calls) + }) + } +} diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index 27a56497a..eca086794 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -29,24 +29,18 @@ import ( _ "github.com/go-sql-driver/mysql" "github.com/uber-go/tally" pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" - "github.com/uber/submitqueue/platform/consumer" - "github.com/uber/submitqueue/platform/errs" - genericerrs "github.com/uber/submitqueue/platform/errs/generic" - mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql" "github.com/uber/submitqueue/platform/extension/consumergate" consumergatefile "github.com/uber/submitqueue/platform/extension/consumergate/file" consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop" "github.com/uber/submitqueue/platform/extension/counter" mysqlcounter "github.com/uber/submitqueue/platform/extension/counter/mysql" - extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" + "github.com/uber/submitqueue/platform/lifecycle" servicemq "github.com/uber/submitqueue/service/messagequeue" "github.com/uber/submitqueue/service/submitqueue/core" "github.com/uber/submitqueue/service/submitqueue/gateway/server/mapper" - "github.com/uber/submitqueue/submitqueue/core/topickey" yamlqueueconfig "github.com/uber/submitqueue/submitqueue/extension/queueconfig/yaml" "github.com/uber/submitqueue/submitqueue/gateway/controller" - logctrl "github.com/uber/submitqueue/submitqueue/gateway/controller/log" requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request" storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" mysqlstorage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mysql" @@ -319,24 +313,7 @@ func run() error { } subscriberName := fmt.Sprintf("%s-%d", hostname, os.Getpid()) - // 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 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}, - { - Key: topickey.TopicKeyLog, - Name: "log", - Queue: mysqlQueue, - Subscription: extqueue.DefaultSubscriptionConfig( - subscriberName, "gateway-log", - ), - }, - }) + registry, err := newGatewayTopicRegistry(mysqlQueue, subscriberName) if err != nil { return fmt.Errorf("failed to create topic registry: %w", err) } @@ -389,32 +366,25 @@ func run() error { // Register reflection service for debugging with grpcurl reflection.Register(grpcServer) - // 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/gateway/extension/storage/mysql) and queue (platform/extension/messagequeue/mysql) - // both run on the same MySQL driver, so a single classifier covers - // errors surfaced from either backend. - genericerrs.Classifier, - mysqlerrs.Classifier, - ), - newConsumerGate(logger), - ) - - logController := logctrl.NewController(logger.Sugar(), scope, materializer, topickey.TopicKeyLog, "gateway-log") - if err := logConsumer.Register(logController); err != nil { - return fmt.Errorf("failed to register log controller: %w", err) + logConsumers, err := newLogConsumers(logger.Sugar(), scope, registry, materializer, newConsumerGate(logger)) + if err != nil { + return err } + return serveGateway(ctx, cancel, grpcServer, logConsumers) +} - if err := logConsumer.Start(ctx); err != nil { - // The error can also be a result of a context cancellation due to SIGINT or SIGTERM. - // This is expected, just propagate it. - return fmt.Errorf("failed to start log consumer: %w", err) +func serveGateway(ctx context.Context, cancel context.CancelFunc, grpcServer *grpc.Server, logConsumers lifecycle.Component) (runErr error) { + if err := logConsumers.Start(ctx); err != nil { + return fmt.Errorf("failed to start log consumers: %w", err) } - logger.Info("log consumer started") + defer func() { + stopErr := logConsumers.Stop(context.Background()) + if stopErr != nil && runErr == context.Canceled { + runErr = stopErr + return + } + runErr = errors.Join(runErr, stopErr) + }() // Listen on configurable port port := os.Getenv("PORT") @@ -426,6 +396,8 @@ func run() error { return fmt.Errorf("failed to listen on port %s: %w", port, err) } + defer listener.Close() + fmt.Printf("Gateway gRPC server is running on %s\n", port) fmt.Println("Press Ctrl+C to stop, or send a SIGTERM.") @@ -435,50 +407,21 @@ func run() error { serverErrCh <- grpcServer.Serve(listener) }() - // Wait for interrupt signal or server critical error - // If interruption is signaled, gracefully stop the server - // If the server exits with an error, cancel the context to signal the consumer - // After this, stop the consumer - // If an error happens during shutdown, return the actual error, not the context cancellation error - var serverErr error + var serverErr, shutdownErr error select { case <-ctx.Done(): - fmt.Println("Shutting down gateway server due to interruption signal...") - - // Set the error to the context cancellation error to be surfaced as a desired exit code by the main function - // to indicate that the server was stopped as intended - // It may be overridden by the server error if any - err = ctx.Err() - - // stop GRPC server and wait for it to exit + shutdownErr = ctx.Err() grpcServer.GracefulStop() serverErr = <-serverErrCh case serverErr = <-serverErrCh: - fmt.Println("Shutting down gateway server due to critical GRPC server error...") - - // Cancel the context to signal cancellation to the queue consumer cancel() } - if serverErr != nil { - serverErr = fmt.Errorf("GRPC server exited with error: %w", serverErr) - } - - // Stop the consumer with a 30s timeout; by this time the context should be - // cancelled and the processing threads may already be exiting; recollect them. - errStop := logConsumer.Stop(30000) - if errStop != nil { - errStop = fmt.Errorf("failed to stop consumer: %w", errStop) - } - - if errStop != nil || serverErr != nil { - // Override context cancellation error with the shutdown error. The server - // error is the primary/root failure, so it leads; the consumer-stop error - // is secondary cleanup. - err = errors.Join(serverErr, errStop) + if serverErr != nil && !errors.Is(serverErr, grpc.ErrServerStopped) { + return fmt.Errorf("GRPC server exited with error: %w", serverErr) } - return err + return shutdownErr } func validateConfiguredQueueTenants(tenants, configuredQueueNames []string) error { diff --git a/submitqueue/gateway/README.md b/submitqueue/gateway/README.md index cd0695524..24eefced5 100644 --- a/submitqueue/gateway/README.md +++ b/submitqueue/gateway/README.md @@ -24,3 +24,11 @@ The gateway owns the request log read model and is the only service that reads i - `GetRequestHistoryByID` and `GetRequestHistoryByChangeURI` read retained request-log rows directly. The materializer appends every audit event, selects the current authoritative winner, and repairs the queue projection. The normal orchestrator pipeline does not read or write the request-log store directly. + +## Final log recovery + +The primary `gateway-log` consumer sends failed deliveries to `log_dlq` after a non-retryable error or exhaustion of its finite retry budget. The gateway's `gateway-log-dlq` consumer replays retained entries through the same log controller and materializer, with unlimited attempts, capped backoff, and no second-level DLQ. Valid log deliveries are acknowledged only after history, summary, URI mappings, and receipt projection succeed. Replays may append equivalent history rows; terminal-summary precedence and mapping repair are unchanged. + +With valid envelope metadata, malformed payloads and payload-queue identity conflicts stay in `log_dlq`, with topic/group-tagged diagnostics and deserialize or identity-error counters. Retrying unchanged bad data does not repair it. This is retention, not isolated quarantine: poison records can occupy the fetch window and delay later recovery. Operators must durably export a bad record and record its disposition before deleting it; corrected payloads must be replayed through gateway materialization. No export/replay CLI is provided. Rolling back the gateway stops automatic recovery but must leave the backlog intact. + +Invalid envelope metadata is logged and rejected before the controller runs. If no further DLQ exists, rejection may acknowledge and discard the delivery without retry; operators investigate the logged failure if needed. diff --git a/submitqueue/gateway/controller/log/log.go b/submitqueue/gateway/controller/log/log.go index d5c87be78..1d8e9fef3 100644 --- a/submitqueue/gateway/controller/log/log.go +++ b/submitqueue/gateway/controller/log/log.go @@ -27,13 +27,7 @@ import ( "go.uber.org/zap" ) -// Controller handles log queue messages. -// It consumes request log entries and persists them to storage. -// Implements consumer.Controller interface for integration with the consumer. -// -// The request log is written exclusively by the gateway: other services -// (e.g. the orchestrator) only publish log entries to the log topic, and this -// controller is the single consumer that persists them to storage. +// Controller materializes gateway-owned request logs from primary or recovery deliveries. type Controller struct { logger *zap.SugaredLogger metricsScope tally.Scope @@ -54,8 +48,8 @@ func NewController( consumerGroup string, ) *Controller { return &Controller{ - logger: logger.Named("log_controller"), - metricsScope: scope.SubScope("log_controller"), + logger: logger.Named("log_controller").With("topic_key", topicKey, "consumer_group", consumerGroup), + metricsScope: scope.SubScope("log_controller").Tagged(map[string]string{"topic_key": topicKey.String(), "consumer_group": consumerGroup}), materializer: materializer, topicKey: topicKey, consumerGroup: consumerGroup, @@ -73,10 +67,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er logEntry, err := sqmq.UnmarshalRequestLog(msg.Payload) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "deserialize_errors", 1) - // Non-retryable: malformed messages will never succeed regardless of retry count return fmt.Errorf("failed to deserialize request log: %w", err) } if err := entityqueue.ValidatePayloadQueue(msg, logEntry.Queue); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "identity_errors", 1) return fmt.Errorf("invalid message identity: %w", err) } @@ -87,13 +81,12 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er "attempt", delivery.Attempt(), ) - // 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 persist request log: %w", err) } - return nil // Success - message will be acked + return nil } // Name returns the controller name for logging and metrics.