feat(gateway): Reject land requests for a paused queue

service/submitqueue/README.md

@@ -21,6 +21,7 @@ Each service requires the application and queue MySQL databases. Service-specifi
 | `MYSQL_DSN` | both | yes | Application database DSN | — |
 | `QUEUE_MYSQL_DSN` | both | yes | Queue database DSN | — |
 | `QUEUE_CONFIG_PATH` | Gateway | yes | YAML registry of accepted queue names | — |
+| `QUEUE_PAUSE_CONFIG_PATH` | Gateway | no | YAML file of per-queue pause switches, re-read on every land | pausing disabled when unset |
 | `PROFILES_CONFIG_PATH` | Orchestrator | no | YAML extension profiles and per-queue routing | built-in example profiles |
 | `HOSTNAME` | both | no | Base subscriber name; Gateway appends its PID | time-based service name |
 | `QUEUE_LOG_LEVEL` | both | no | Message-queue logger level | `info` |

service/submitqueue/gateway/server/BUILD.bazel

@@ -32,6 +32,9 @@ go_library(
         "//service/submitqueue/gateway/server/mapper:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/extension/queueconfig/yaml:go_default_library",
+        "//submitqueue/extension/queuepause:go_default_library",
+        "//submitqueue/extension/queuepause/file:go_default_library",
+        "//submitqueue/extension/queuepause/noop:go_default_library",
         "//submitqueue/gateway/controller:go_default_library",
         "//submitqueue/gateway/controller/log:go_default_library",
         "//submitqueue/gateway/core/request:go_default_library",
@@ -86,6 +89,7 @@ go_test(
     embed = [":gateway_lib"],  # keep
     deps = [
         "//api/submitqueue/gateway/protopb:go_default_library",
+        "//platform/errs:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/queueconfig/mock:go_default_library",
         "//submitqueue/gateway/controller:go_default_library",

service/submitqueue/gateway/server/main.go

@@ -44,6 +44,9 @@ import (
 	"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/extension/queuepause"
+	queuepausefile "github.com/uber/submitqueue/submitqueue/extension/queuepause/file"
+	queuepausenoop "github.com/uber/submitqueue/submitqueue/extension/queuepause/noop"
 	"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"
@@ -149,6 +152,8 @@ func (s *GatewayServer) GetRequestHistoryByChangeURI(ctx context.Context, req *p
 	return &pb.GetRequestHistoryByChangeURIResponse{Histories: mapper.RequestHistoriesToProto(histories)}, nil
 }
 
+// gatewayStatusError maps controller errors to status codes. IsInvalidRequest
+// matches every user error, so each specific user error is checked before it.
 func gatewayStatusError(err error) error {
 	switch {
 	case controller.IsRequestNotFound(err):
@@ -157,13 +162,31 @@ func gatewayStatusError(err error) error {
 		return status.Error(codes.ResourceExhausted, err.Error())
 	case controller.IsInternalConsistency(err):
 		return status.Error(codes.Internal, err.Error())
+	case controller.IsQueuePaused(err):
+		return status.Error(codes.FailedPrecondition, err.Error())
 	case controller.IsInvalidRequest(err), controller.IsUnrecognizedQueue(err):
 		return status.Error(codes.InvalidArgument, err.Error())
 	default:
 		return err
 	}
 }
 
+// newQueuePauseStore enables the file-backed pause store only when
+// QUEUE_PAUSE_CONFIG_PATH is set; otherwise no queue is ever paused.
+func newQueuePauseStore(logger *zap.Logger) (queuepause.Store, error) {
+	path := os.Getenv("QUEUE_PAUSE_CONFIG_PATH")
+	if path == "" {
+		logger.Info("queue pause disabled")
+		return queuepausenoop.New(), nil
+	}
+	store, err := queuepausefile.NewStore(path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load queue pause config: %w", err)
+	}
+	logger.Info("queue pause configured", zap.String("path", path))
+	return store, nil
+}
+
 func main() {
 	code := 0
 	if err := run(); err != nil {
@@ -264,6 +287,10 @@ func run() error {
 	if err != nil {
 		return fmt.Errorf("failed to load queue configs: %w", err)
 	}
+	queuePauses, err := newQueuePauseStore(logger)
+	if err != nil {
+		return err
+	}
 	configuredQueues, err := queueConfigs.List(ctx)
 	if err != nil {
 		return fmt.Errorf("failed to list queue configs: %w", err)
@@ -359,7 +386,7 @@ func run() error {
 	storageFty := storageFactory{backend: store}
 	materializer := requestcore.NewMaterializer(storageFty)
 	pingController := controller.NewPingController(logger, scope)
-	landController := controller.NewLandController(logger.Sugar(), scope, cnt, storageFty, materializer, queueConfigs, registry)
+	landController := controller.NewLandController(logger.Sugar(), scope, cnt, storageFty, materializer, queueConfigs, queuePauses, registry)
 	cancelController := controller.NewCancelController(logger.Sugar(), scope, storageFty, materializer, registry)
 	requestSummaryController := controller.NewRequestSummaryController(logger.Sugar(), scope, storageFty)
 	listController := controller.NewListController(logger.Sugar(), scope, storageFty, queueConfigs)

service/submitqueue/gateway/server/main_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally"
 	pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb"
+	"github.com/uber/submitqueue/platform/errs"
 	"github.com/uber/submitqueue/submitqueue/entity"
 	qcmock "github.com/uber/submitqueue/submitqueue/extension/queueconfig/mock"
 	"github.com/uber/submitqueue/submitqueue/gateway/controller"
@@ -102,9 +103,14 @@ func TestGatewayStatusError(t *testing.T) {
 		},
 		{
 			name: "unrecognized queue",
-			err:  &controller.UnrecognizedQueueError{Queue: "missing"},
+			err:  errs.NewUserError(&controller.UnrecognizedQueueError{Queue: "missing"}),
 			code: codes.InvalidArgument,
 		},
+		{
+			name: "queue paused",
+			err:  errs.NewUserError(&controller.QueuePausedError{Queue: "paused"}),
+			code: codes.FailedPrecondition,
+		},
 		{
 			name: "request not found",
 			err:  &controller.RequestNotFoundError{Sqid: "1"},

submitqueue/gateway/controller/BUILD.bazel

@@ -26,6 +26,7 @@ go_library(
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/queueconfig:go_default_library",
+        "//submitqueue/extension/queuepause:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/gateway/core/request:go_default_library",
         "//submitqueue/gateway/extension/storage:go_default_library",
@@ -63,6 +64,9 @@ go_test(
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/queueconfig:go_default_library",
         "//submitqueue/extension/queueconfig/mock:go_default_library",
+        "//submitqueue/extension/queuepause:go_default_library",
+        "//submitqueue/extension/queuepause/fake:go_default_library",
+        "//submitqueue/extension/queuepause/noop:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/extension/storage/mock:go_default_library",
         "//submitqueue/gateway/core/request:go_default_library",

submitqueue/gateway/controller/land.go

@@ -31,6 +31,7 @@ import (
 	"github.com/uber/submitqueue/submitqueue/core/topickey"
 	"github.com/uber/submitqueue/submitqueue/entity"
 	"github.com/uber/submitqueue/submitqueue/extension/queueconfig"
+	"github.com/uber/submitqueue/submitqueue/extension/queuepause"
 	requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request"
 	storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage"
 	"go.uber.org/zap"
@@ -63,6 +64,28 @@ func IsUnrecognizedQueue(err error) bool {
 	return errors.As(err, &target)
 }
 
+// QueuePausedError indicates the queue is refusing new land requests.
+type QueuePausedError struct {
+	Queue string
+	// Reason is the operator's explanation for the pause, empty when none was given.
+	Reason string
+}
+
+// Error implements the error interface.
+func (e *QueuePausedError) Error() string {
+	msg := fmt.Sprintf("queue %q is paused and not accepting land requests", e.Queue)
+	if e.Reason != "" {
+		msg += ": " + e.Reason
+	}
+	return msg
+}
+
+// IsQueuePaused returns true if any error in the chain is a *QueuePausedError.
+func IsQueuePaused(err error) bool {
+	var target *QueuePausedError
+	return errors.As(err, &target)
+}
+
 // counterDomainRequest names the per-queue sequence that mints request IDs.
 const counterDomainRequest = "request"
 
@@ -80,20 +103,22 @@ type landController struct {
 	stores       storage.Factory
 	materializer *requestcore.Materializer
 	queueConfigs queueconfig.Store
+	queuePauses  queuepause.Store
 	registry     consumer.TopicRegistry
 }
 
 // NewLandController creates a new instance of the gateway land controller.
 // The controller publishes land requests to the topic registered under
 // topickey.TopicKeyStart in the registry.
-func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counters counter.Factory, stores storage.Factory, materializer *requestcore.Materializer, queueConfigs queueconfig.Store, registry consumer.TopicRegistry) LandController {
+func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counters counter.Factory, stores storage.Factory, materializer *requestcore.Materializer, queueConfigs queueconfig.Store, queuePauses queuepause.Store, registry consumer.TopicRegistry) LandController {
 	return &landController{
 		logger:       logger,
 		metricsScope: scope.SubScope("land_controller"),
 		counters:     counters,
 		stores:       stores,
 		materializer: materializer,
 		queueConfigs: queueConfigs,
+		queuePauses:  queuePauses,
 		registry:     registry,
 	}
 }
@@ -121,6 +146,15 @@ func (c *landController) Land(ctx context.Context, req entity.LandRequest) (resu
 		return entity.LandResult{}, fmt.Errorf("failed to look up queue %q: %w", queue, err)
 	}
 
+	// A failed read fails the land: an unreadable pause state must not let a paused queue accept work.
+	pauseState, err := c.queuePauses.Get(ctx, queue)
+	if err != nil {
+		return entity.LandResult{}, fmt.Errorf("failed to read pause state for queue=%s: %w", queue, err)
+	}
+	if pauseState.RejectLands {
+		return entity.LandResult{}, errs.NewUserError(&QueuePausedError{Queue: queue, Reason: pauseState.Reason})
+	}
+
 	// Generate a queue-scoped request ID for the land request.
 	// The inbound entity arrives with an empty ID; the controller owns minting it.
 	stores, err := c.stores.For(storage.Config{QueueName: queue})

submitqueue/gateway/controller/land_test.go

@@ -36,6 +36,9 @@ import (
 	"github.com/uber/submitqueue/submitqueue/entity"
 	"github.com/uber/submitqueue/submitqueue/extension/queueconfig"
 	qcmock "github.com/uber/submitqueue/submitqueue/extension/queueconfig/mock"
+	"github.com/uber/submitqueue/submitqueue/extension/queuepause"
+	queuepausefake "github.com/uber/submitqueue/submitqueue/extension/queuepause/fake"
+	queuepausenoop "github.com/uber/submitqueue/submitqueue/extension/queuepause/noop"
 	storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock"
 	requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request"
 	gwstoragemock "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mock"
@@ -87,7 +90,7 @@ func (f staticCounterFactory) For(config counter.Config) (counter.Counter, error
 // noopStorage returns stateful request storage whose writes succeed.
 func newNoopLandController(t *testing.T, ctrl *gomock.Controller, cnt counter.Counter) LandController {
 	fixture := newControllerStorageFixture(ctrl)
-	return NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), noopQueueConfigStore(ctrl), newTestRegistryWithNoopPublisher(t, ctrl))
+	return NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), noopQueueConfigStore(ctrl), queuepausenoop.New(), newTestRegistryWithNoopPublisher(t, ctrl))
 }
 
 // noopQueueConfigStore returns a mock queueconfig.Store that always reports
@@ -178,6 +181,7 @@ func TestLand_ResolvesCounterForRequestQueue(t *testing.T) {
 		fixture.newFactory(ctrl),
 		fixture.newMaterializer(ctrl),
 		noopQueueConfigStore(ctrl),
+		queuepausenoop.New(),
 		newTestRegistryWithNoopPublisher(t, ctrl),
 	)
 	ctx := context.Background()
@@ -307,7 +311,7 @@ func TestLand_ReturnsUnrecognizedQueueWhenStoreReportsNotFound(t *testing.T) {
 	qcs.EXPECT().Get(gomock.Any(), "missing-queue").Return(entity.QueueConfig{}, queueconfig.ErrNotFound)
 
 	fixture := newControllerStorageFixture(ctrl)
-	controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, newTestRegistryWithNoopPublisher(t, ctrl))
+	controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, queuepausenoop.New(), newTestRegistryWithNoopPublisher(t, ctrl))
 	ctx := context.Background()
 
 	_, err := controller.Land(ctx, testLandRequest("missing-queue"))
@@ -330,7 +334,7 @@ func TestLand_PropagatesQueueConfigStoreError(t *testing.T) {
 	qcs.EXPECT().Get(gomock.Any(), "test-queue").Return(entity.QueueConfig{}, fmt.Errorf("config backend down"))
 
 	fixture := newControllerStorageFixture(ctrl)
-	controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, newTestRegistryWithNoopPublisher(t, ctrl))
+	controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, queuepausenoop.New(), newTestRegistryWithNoopPublisher(t, ctrl))
 	ctx := context.Background()
 
 	_, err := controller.Land(ctx, testLandRequest("test-queue"))
@@ -340,6 +344,81 @@ func TestLand_PropagatesQueueConfigStoreError(t *testing.T) {
 	assert.False(t, IsInvalidRequest(err))
 }
 
+func newLandControllerWithPauseStore(t *testing.T, ctrl *gomock.Controller, cnt counter.Counter, qcs *qcmock.MockStore, pauses queuepause.Store) LandController {
+	fixture := newControllerStorageFixture(ctrl)
+	return NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, pauses, newTestRegistryWithNoopPublisher(t, ctrl))
+}
+
+func TestLand_QueuePause(t *testing.T) {
+	tests := []struct {
+		name       string
+		state      entity.QueuePause
+		wantLanded bool
+		wantReason string
+	}{
+		{name: "not paused", state: entity.QueuePause{}, wantLanded: true},
+		{name: "freeze only still accepts lands", state: entity.QueuePause{FreezeProcessing: true}, wantLanded: true},
+		{name: "reject only", state: entity.QueuePause{RejectLands: true}, wantLanded: false},
+		{name: "reject and freeze", state: entity.QueuePause{RejectLands: true, FreezeProcessing: true}, wantLanded: false},
+		{name: "reject carries the reason", state: entity.QueuePause{RejectLands: true, Reason: "repo freeze until 17:00"}, wantLanded: false, wantReason: "repo freeze until 17:00"},
+		{name: "reason alone does not reject", state: entity.QueuePause{Reason: "stale note"}, wantLanded: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			cnt := countermock.NewMockCounter(ctrl)
+			if tt.wantLanded {
+				cnt.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(1), nil)
+			}
+			pauses := queuepausefake.New().Set("test-queue", tt.state)
+			controller := newLandControllerWithPauseStore(t, ctrl, cnt, noopQueueConfigStore(ctrl), pauses)
+
+			result, err := controller.Land(context.Background(), testLandRequest("test-queue"))
+
+			if tt.wantLanded {
+				require.NoError(t, err)
+				assert.Equal(t, "1", result.ID)
+				return
+			}
+			require.Error(t, err)
+			assert.True(t, IsQueuePaused(err))
+			assert.True(t, errs.IsUserError(err))
+			assert.False(t, errs.IsRetryable(err))
+			var typed *QueuePausedError
+			require.ErrorAs(t, err, &typed)
+			assert.Equal(t, "test-queue", typed.Queue)
+			assert.Equal(t, tt.wantReason, typed.Reason)
+		})
+	}
+}
+
+func TestLand_FailsClosedWhenPauseStoreErrors(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	cnt := countermock.NewMockCounter(ctrl)
+	pauses := queuepausefake.New().FailWith(fmt.Errorf("pause backend down"))
+	controller := newLandControllerWithPauseStore(t, ctrl, cnt, noopQueueConfigStore(ctrl), pauses)
+
+	_, err := controller.Land(context.Background(), testLandRequest("test-queue"))
+
+	require.Error(t, err)
+	assert.False(t, IsQueuePaused(err))
+	assert.False(t, errs.IsUserError(err))
+}
+
+func TestLand_ReportsUnrecognizedQueueBeforeReadingPauseState(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	cnt := countermock.NewMockCounter(ctrl)
+	qcs := qcmock.NewMockStore(ctrl)
+	qcs.EXPECT().Get(gomock.Any(), "missing-queue").Return(entity.QueueConfig{}, queueconfig.ErrNotFound)
+	pauses := queuepausefake.New().FailWith(fmt.Errorf("pause backend down"))
+	controller := newLandControllerWithPauseStore(t, ctrl, cnt, qcs, pauses)
+
+	_, err := controller.Land(context.Background(), testLandRequest("missing-queue"))
+
+	require.Error(t, err)
+	assert.True(t, IsUnrecognizedQueue(err))
+}
+
 func TestLand_PublishesToQueue(t *testing.T) {
 	var publishedTopic string
 	var publishedMessage entityqueue.Message
@@ -414,7 +493,7 @@ func TestLand_PublishesToQueue(t *testing.T) {
 		),
 	)
 
-	controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, store), materializer, noopQueueConfigStore(ctrl), registry)
+	controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, store), materializer, noopQueueConfigStore(ctrl), queuepausenoop.New(), registry)
 	ctx := context.Background()
 
 	req := entity.LandRequest{
@@ -486,7 +565,7 @@ func TestLand_ReturnsErrorWhenPublishFails(t *testing.T) {
 	registry, publisher := newTestRegistry(t, ctrl)
 	publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("queue unavailable"))
 
-	controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, storageWithSummaryStore(ctrl, summaryStore)), newControllerStorageFixture(ctrl).newMaterializer(ctrl), noopQueueConfigStore(ctrl), registry)
+	controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, storageWithSummaryStore(ctrl, summaryStore)), newControllerStorageFixture(ctrl).newMaterializer(ctrl), noopQueueConfigStore(ctrl), queuepausenoop.New(), registry)
 	ctx := context.Background()
 
 	_, err := controller.Land(ctx, testLandRequest("test-queue"))
@@ -508,6 +587,7 @@ func TestLand_ReturnsSqidWhenAcceptedLogFailsAfterPublish(t *testing.T) {
 		fixture.newFactory(ctrl),
 		fixture.newMaterializer(ctrl),
 		noopQueueConfigStore(ctrl),
+		queuepausenoop.New(),
 		newTestRegistryWithNoopPublisher(t, ctrl),
 	)
 	result, err := controller.Land(context.Background(), testLandRequest("test-queue"))