feat(consumergate): runtime stop/start of queue controllers + deterministic e2e cancel test

Implement doc/rfc/consumer-gate.md:

- platform/extension/consumergate: extension contract — Gate, a single
  blocking Wait the consumer consults per delivery (implementations own the
  wait mechanism, so a notification-capable store can release instantly, and
  the parked-delivery observation records), Admin (write surface for tests
  and tooling), Config, Factory interface, and gomock mocks.
- platform/extension/consumergate/file: polling implementation — gate state
  as plain files in a shared directory (gate file present = closed, rm =
  open), parked-delivery records as JSON stamped parked/released by the
  store, per-partition cached verdicts (TTL = poll interval) so an open gate
  costs no stat per message, and temp-file-plus-rename writes with converged
  error handling.
- platform/extension/consumergate/noop: no-op gate for services and tests
  that do not need runtime gating.
- platform/consumer: consumer.New takes the Gate as a required argument and
  consults it before each delivery. While a delivery is blocked the consumer
  keeps it in-flight by periodically extending visibility (no retry budget
  burned, partition order preserved). Shutdown while blocked leaves the
  delivery for normal redelivery; gate errors fail open with a log and
  counter.
- Wiring: gateway, orchestrator (primary + DLQ), and runway construct the
  file-backed gate rooted at CONSUMER_GATE_DIR, defaulting to
  /var/submitqueue/consumergate — the path the compose stack bind-mounts
  into every service; stovepipe wires the no-op gate.
- test/e2e/submitqueue: replace TestCancel_RecordsIntent with
  TestCancel_CaughtPreBatch_NeverLands — the stop→observe→start scenario from
  the RFC. The gate parks runway's merge-conflict check for the test queue's
  partition before landing, the parked record proves the request is held
  pre-batch, the cancel drives it terminal cancelled, and after the gate
  opens a sentinel request landing on the same partitions proves the stale
  check signal was consumed and dropped — the cancelled change is never
  batched and never lands.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: sbalabanov-zz

Makefile

@@ -364,7 +364,7 @@ local-stovepipe-stop: ## Stop the Stovepipe service
 
 mocks: ## Generate mock files using mockgen
 	@echo "Generating mocks..."
-	@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/pusher/... ./submitqueue/extension/scorer/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/enumerator/... ./submitqueue/extension/speculation/dependencylimit/... ./submitqueue/extension/speculation/selector/... ./submitqueue/extension/speculation/selectionlimit/... ./submitqueue/extension/speculation/prioritizer/... ./submitqueue/extension/speculation/prioritizationlimit/... ./submitqueue/extension/validator/... ./submitqueue/extension/speculation/pathscorer/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
+	@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/pusher/... ./submitqueue/extension/scorer/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/enumerator/... ./submitqueue/extension/speculation/dependencylimit/... ./submitqueue/extension/speculation/selector/... ./submitqueue/extension/speculation/selectionlimit/... ./submitqueue/extension/speculation/prioritizer/... ./submitqueue/extension/speculation/prioritizationlimit/... ./submitqueue/extension/validator/... ./submitqueue/extension/speculation/pathscorer/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
 	@echo "Mocks generated successfully!"
 
 proto: ## Generate protobuf files from .proto definitions

platform/consumer/BUILD.bazel

@@ -5,13 +5,15 @@ go_library(
     srcs = [
         "consumer.go",
         "controller.go",
+        "gate.go",
         "registry.go",
     ],
     importpath = "github.com/uber/submitqueue/platform/consumer",
     visibility = ["//visibility:public"],
     deps = [
         "//platform/base/messagequeue:go_default_library",
         "//platform/errs:go_default_library",
+        "//platform/extension/consumergate:go_default_library",
         "//platform/extension/messagequeue:go_default_library",
         "//platform/metrics:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
@@ -23,13 +25,17 @@ go_test(
     name = "go_default_test",
     srcs = [
         "consumer_test.go",
+        "gate_internal_test.go",
+        "gate_test.go",
         "registry_test.go",
     ],
+    embed = [":go_default_library"],
     deps = [
-        ":go_default_library",
         "//platform/base/messagequeue:go_default_library",
         "//platform/consumer/mock:go_default_library",
         "//platform/errs:go_default_library",
+        "//platform/extension/consumergate:go_default_library",
+        "//platform/extension/consumergate/noop:go_default_library",
         "//platform/extension/messagequeue:go_default_library",
         "//platform/extension/messagequeue/mock:go_default_library",
         "//submitqueue/core/topickey:go_default_library",

platform/consumer/consumer.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/uber-go/tally"
 	"github.com/uber/submitqueue/platform/errs"
+	"github.com/uber/submitqueue/platform/extension/consumergate"
 	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
 	"github.com/uber/submitqueue/platform/metrics"
 	"go.uber.org/zap"
@@ -62,6 +63,7 @@ type consumer struct {
 	metricsScope tally.Scope
 	registry     TopicRegistry
 	processor    errs.ErrorProcessor
+	gate         consumergate.Gate
 
 	mu            sync.Mutex
 	stopped       bool
@@ -86,12 +88,18 @@ type activeSubscription struct {
 // consumers such as DLQ reconciliation that must redeliver on any failure.
 // processor must not be nil; callers that genuinely want no transformation
 // can pass errs.NewClassifierProcessor() with no classifiers.
-func New(logger *zap.SugaredLogger, scope tally.Scope, registry TopicRegistry, processor errs.ErrorProcessor) Consumer {
+//
+// gate is the consumer-gate implementation consulted before each delivery
+// reaches its controller. Pass noop.New() (from
+// platform/extension/consumergate/noop) for services that do not need runtime
+// gating. gate must not be nil.
+func New(logger *zap.SugaredLogger, scope tally.Scope, registry TopicRegistry, processor errs.ErrorProcessor, gate consumergate.Gate) Consumer {
 	return &consumer{
 		logger:        logger,
 		metricsScope:  scope.SubScope("consumer"),
 		registry:      registry,
 		processor:     processor,
+		gate:          gate,
 		subscriptions: make(map[TopicKey]*activeSubscription),
 	}
 }
@@ -342,6 +350,14 @@ func (m *consumer) processPartition(ctx context.Context, controller Controller,
 func (m *consumer) processDelivery(ctx context.Context, controller Controller, delivery extqueue.Delivery, controllerScope tally.Scope) {
 	const opName = "process"
 
+	// Consumer gate: block the delivery while the controller's gate is closed.
+	// A false return means the consumer is shutting down while blocked — leave
+	// the delivery in-flight (no process, no ack/nack) so its visibility lapses
+	// into a normal redelivery. Gate errors fail open inside waitGate.
+	if !m.waitGate(ctx, controller, delivery, controllerScope) {
+		return
+	}
+
 	start := time.Now()
 	metrics.NamedCounter(controllerScope, opName, "messages_received", 1)
 

platform/consumer/consumer_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/uber/submitqueue/platform/consumer"
 	consumermock "github.com/uber/submitqueue/platform/consumer/mock"
 	"github.com/uber/submitqueue/platform/errs"
+	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/submitqueue/core/topickey"
@@ -92,7 +93,7 @@ func TestNew(t *testing.T) {
 	reg, err := consumer.NewTopicRegistry(nil)
 	require.NoError(t, err)
 
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 	require.NotNil(t, c)
 }
 
@@ -101,7 +102,7 @@ func TestConsumer_Register(t *testing.T) {
 	logger := zaptest.NewLogger(t).Sugar()
 
 	reg, _ := consumer.NewTopicRegistry(nil)
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	handler1 := consumermock.NewMockController(ctrl)
 	setupController(handler1, "handler1", topickey.TopicKeyStart, "group1", nil)
@@ -121,7 +122,7 @@ func TestConsumer_Register_DuplicateTopic(t *testing.T) {
 	logger := zaptest.NewLogger(t).Sugar()
 
 	reg, _ := consumer.NewTopicRegistry(nil)
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	handler1 := consumermock.NewMockController(ctrl)
 	setupController(handler1, "handler1", topickey.TopicKeyStart, "group1", nil)
@@ -141,7 +142,7 @@ func TestConsumer_Register_AfterStop(t *testing.T) {
 	logger := zaptest.NewLogger(t).Sugar()
 
 	reg, _ := consumer.NewTopicRegistry(nil)
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	err := c.Stop(1000)
 	require.NoError(t, err)
@@ -157,7 +158,7 @@ func TestConsumer_Start_NoHandlers(t *testing.T) {
 	logger := zaptest.NewLogger(t).Sugar()
 
 	reg, _ := consumer.NewTopicRegistry(nil)
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	err := c.Start(context.Background())
 	assert.Error(t, err)
@@ -168,7 +169,7 @@ func TestConsumer_Start_AfterStop(t *testing.T) {
 	logger := zaptest.NewLogger(t).Sugar()
 
 	reg, _ := consumer.NewTopicRegistry(nil)
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	handler := consumermock.NewMockController(ctrl)
 	setupController(handler, "handler1", topickey.TopicKeyStart, "group1", nil)
@@ -194,7 +195,7 @@ func TestConsumer_Start_MissingSubscriptionConfig(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	handler := consumermock.NewMockController(ctrl)
 	setupController(handler, "handler", topickey.TopicKeyStart, "group", nil)
@@ -220,7 +221,7 @@ func TestConsumer_Start_SubscribeFailure(t *testing.T) {
 
 	reg := newRegistry(t, mockQ, topickey.TopicKeyStart, "group")
 
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	handler := consumermock.NewMockController(ctrl)
 	setupController(handler, "handler", topickey.TopicKeyStart, "group", nil)
@@ -246,7 +247,7 @@ func TestConsumer_ProcessDelivery_Success(t *testing.T) {
 
 	reg := newRegistry(t, mockQ, topickey.TopicKeyStart, "test-group")
 
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	handledMsg := ""
 	handler := consumermock.NewMockController(ctrl)
@@ -292,7 +293,7 @@ func TestConsumer_ProcessDelivery_Error(t *testing.T) {
 
 	reg := newRegistry(t, mockQ, topickey.TopicKeyStart, "test-group")
 
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	handler := consumermock.NewMockController(ctrl)
 	setupController(handler, "test-handler", topickey.TopicKeyStart, "test-group",
@@ -334,7 +335,7 @@ func TestConsumer_ProcessDelivery_NonRetryableError(t *testing.T) {
 
 	reg := newRegistry(t, mockQ, topickey.TopicKeyStart, "test-group")
 
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	handler := consumermock.NewMockController(ctrl)
 	setupController(handler, "test-handler", topickey.TopicKeyStart, "test-group",
@@ -385,7 +386,7 @@ func TestConsumer_Stop(t *testing.T) {
 
 	reg := newRegistry(t, mockQ, topickey.TopicKeyStart, "test-group")
 
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	handler := consumermock.NewMockController(ctrl)
 	setupController(handler, "test-handler", topickey.TopicKeyStart, "test-group", nil)
@@ -443,7 +444,7 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 
 			reg := newRegistry(t, mockQ, topickey.TopicKeyStart, "test-group")
 
-			testC := consumer.New(logger, testScope, reg, errs.NewClassifierProcessor())
+			testC := consumer.New(logger, testScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 			handler := consumermock.NewMockController(ctrl)
 			setupController(handler, "test-handler", topickey.TopicKeyStart, "test-group",
@@ -518,7 +519,7 @@ func TestConsumer_AckNackLatencyTracking(t *testing.T) {
 
 	reg := newRegistry(t, mockQ, topickey.TopicKeyStart, "test-group")
 
-	c := consumer.New(logger, scope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, scope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	handler := consumermock.NewMockController(ctrl)
 	setupController(handler, "test-handler", topickey.TopicKeyStart, "test-group",
@@ -563,7 +564,7 @@ func TestConsumer_ErrorMetrics(t *testing.T) {
 
 	reg := newRegistry(t, mockQ, topickey.TopicKeyStart, "test-group")
 
-	c := consumer.New(logger, scope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, scope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	handler := consumermock.NewMockController(ctrl)
 	setupController(handler, "test-handler", topickey.TopicKeyStart, "test-group",
@@ -619,7 +620,7 @@ func TestConsumer_PerPartitionProcessing(t *testing.T) {
 
 	reg := newRegistry(t, mockQ, topickey.TopicKeyStart, "test-group")
 
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	// Track processing by partition
 	partBDone := make(chan struct{})
@@ -704,7 +705,7 @@ func TestConsumer_PartitionOrdering(t *testing.T) {
 
 	reg := newRegistry(t, mockQ, topickey.TopicKeyStart, "test-group")
 
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	// Mutex + shared slice captures processing order for assertion;
 	// a channel would only signal completion, not record the sequence.
@@ -773,7 +774,7 @@ func TestConsumer_PartitionWorkerCleanup(t *testing.T) {
 
 	reg := newRegistry(t, mockQ, topickey.TopicKeyStart, "test-group")
 
-	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor())
+	c := consumer.New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	processedCount := int64(0)
 

platform/consumer/gate.go

@@ -0,0 +1,125 @@
+// Copyright (c) 2026 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 consumer
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/platform/extension/consumergate"
+	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
+	"github.com/uber/submitqueue/platform/metrics"
+)
+
+const (
+	// parkExtensionMs is the visibility extension applied to a parked delivery
+	// on each keep-in-flight tick, keeping it in-flight without burning retry
+	// budget (milliseconds). Must comfortably exceed the extension cadence.
+	parkExtensionMs = int64(30000)
+
+	// keepInFlightInterval is how often a blocked delivery's visibility is
+	// extended while the gate holds it.
+	keepInFlightInterval = 10 * time.Second
+)
+
+// waitGate consults the consumer gate before letting a delivery reach the
+// controller. It returns true when the delivery may proceed, false when the
+// consumer is shutting down while blocked (the delivery is left in-flight for
+// redelivery). Gate errors fail open: the delivery proceeds with a logged
+// warning.
+func (m *consumer) waitGate(ctx context.Context, controller Controller, delivery extqueue.Delivery, scope tally.Scope) bool {
+	const opName = "gate"
+
+	msg := delivery.Message()
+	parked := consumergate.Parked{
+		ConsumerGroup: controller.ConsumerGroup(),
+		Topic:         controller.TopicKey().String(),
+		MessageID:     msg.ID,
+		PartitionKey:  msg.PartitionKey,
+		Payload:       msg.Payload,
+		Attempt:       delivery.Attempt(),
+	}
+
+	stopKeepAlive := m.keepInFlight(ctx, delivery, keepInFlightInterval)
+	start := time.Now()
+	err := m.gate.Wait(ctx, parked)
+	stopKeepAlive()
+	metrics.NamedTimer(scope, opName, "wait_latency", time.Since(start))
+
+	if err == nil {
+		return true
+	}
+
+	if ctx.Err() != nil {
+		metrics.NamedCounter(scope, opName, "shutdown_while_blocked", 1)
+		m.logger.Infow("consumer shutdown while delivery was blocked by gate",
+			"consumer_group", parked.ConsumerGroup,
+			"topic", parked.Topic,
+			"message_id", msg.ID,
+		)
+		return false
+	}
+
+	// Any other error: fail open — let the delivery through.
+	metrics.NamedCounter(scope, opName, "wait_errors", 1)
+	m.logger.Errorw("gate wait failed, failing open",
+		"consumer_group", parked.ConsumerGroup,
+		"topic", parked.Topic,
+		"message_id", msg.ID,
+		"error", err,
+	)
+	return true
+}
+
+// keepInFlight starts a goroutine that periodically extends a delivery's
+// visibility timeout while it is blocked by the gate, so the queue does not
+// redeliver a parked message. The returned stop function joins the goroutine
+// before returning, guaranteeing no extension races the controller.
+func (m *consumer) keepInFlight(ctx context.Context, delivery extqueue.Delivery, interval time.Duration) (stop func()) {
+	var wg sync.WaitGroup
+	done := make(chan struct{})
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-done:
+				return
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				if err := delivery.ExtendVisibilityTimeout(ctx, parkExtensionMs); err != nil {
+					m.logger.Errorw("failed to extend visibility of parked delivery",
+						"message_id", delivery.Message().ID,
+						"error", err,
+					)
+				}
+			}
+		}
+	}()
+
+	var once sync.Once
+	return func() {
+		once.Do(func() {
+			close(done)
+			wg.Wait()
+		})
+	}
+}