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

Implement doc/rfc/consumer-gate.md:

- platform/extension/consumergate: extension contract — Gate.Enter checks a
  delivery against its gate synchronously and returns an Entry future: an
  open gate admits the delivery immediately, a closed gate hands out a
  blocked Entry whose Wait records the parked delivery and blocks until the
  gate opens (implementations own the wait mechanism, so a
  notification-capable store can release instantly, and the parked-delivery
  observation records). Admin is the write surface for tests and tooling;
  Config, Factory interface, and gomock mocks round out the contract.
- 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 delivery, 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
  clears every delivery through Enter. Only a blocked delivery pays for
  queue mechanics: a lazily armed timer keeps it in-flight by periodically
  extending visibility (no goroutine unless actually blocked, no retry
  budget burned, partition order preserved), and blocked wait time is
  recorded on a latency histogram. Shutdown while blocked leaves the
  delivery for normal redelivery; gate errors and failed visibility
  extensions cancel the wait and 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

@@ -12,6 +12,7 @@ go_library(
     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",
@@ -22,14 +23,17 @@ go_library(
 go_test(
     name = "go_default_test",
     srcs = [
+        "consumer_internal_test.go",
         "consumer_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"
@@ -32,6 +33,16 @@ const (
 	// startupCleanupTimeoutMs is the timeout for cleaning up subscriptions when
 	// a controller fails to start during Start().
 	startupCleanupTimeoutMs = 30000
+
+	// gateExtensionMs is the visibility extension applied to a delivery blocked
+	// behind its consumer gate on each keep-in-flight tick, keeping it in-flight
+	// without burning retry budget (milliseconds). Must comfortably exceed
+	// defaultGateExtendInterval.
+	gateExtensionMs = int64(30000)
+
+	// defaultGateExtendInterval is how often a gate-blocked delivery's
+	// visibility is extended.
+	defaultGateExtendInterval = 10 * time.Second
 )
 
 // Consumer orchestrates multiple queue consumers. It handles subscription lifecycle,
@@ -61,6 +72,12 @@ type consumer struct {
 	metricsScope tally.Scope
 	registry     TopicRegistry
 	processor    errs.ErrorProcessor
+	gate         consumergate.Gate
+
+	// gateExtendInterval is how often a gate-blocked delivery's visibility is
+	// extended. Fixed to defaultGateExtendInterval by New; a field (not the
+	// const) so in-package tests can exercise the keep-in-flight path quickly.
+	gateExtendInterval time.Duration
 
 	mu            sync.Mutex
 	stopped       bool
@@ -85,13 +102,20 @@ 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,
-		subscriptions: make(map[TopicKey]*activeSubscription),
+		logger:             logger,
+		metricsScope:       scope.SubScope("consumer"),
+		registry:           registry,
+		processor:          processor,
+		gate:               gate,
+		gateExtendInterval: defaultGateExtendInterval,
+		subscriptions:      make(map[TopicKey]*activeSubscription),
 	}
 }
 
@@ -341,6 +365,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)
 
@@ -485,6 +517,146 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 	)
 }
 
+// waitGate clears a delivery through the consumer gate before it reaches the
+// controller. It returns true when the delivery may proceed, false when it
+// must be dropped without processing or ack/nack (the visibility timeout then
+// lapses into a normal redelivery).
+//
+// Gate.Enter checks the gate synchronously; an unblocked entry is the common
+// path and costs nothing further. Only for a blocked entry does the consumer
+// arrange its queue mechanics — a keep-in-flight timer periodically extending
+// the delivery's visibility — before blocking on the entry's Wait. Failures
+// fail open: if gate state cannot be read or recorded, or the delivery can no
+// longer be held safely because a visibility extension failed, the delivery
+// proceeds and the failure is surfaced via log and counter. Only consumer
+// shutdown drops the delivery.
+func (m *consumer) waitGate(ctx context.Context, controller Controller, delivery extqueue.Delivery, scope tally.Scope) bool {
+	const opName = "gate"
+
+	msg := delivery.Message()
+	consumerGroup := controller.ConsumerGroup()
+	topic := controller.TopicKey().String()
+
+	entry, err := m.gate.Enter(ctx, consumergate.Key{ConsumerGroup: consumerGroup, PartitionKey: msg.PartitionKey})
+	if err != nil {
+		// Gate state could not be read: fail open — gating is auxiliary, and
+		// a broken gate medium must not become a pipeline stall.
+		metrics.NamedCounter(scope, opName, "enter_errors", 1)
+		m.logger.Errorw("gate check failed, failing open",
+			"consumer_group", consumerGroup,
+			"topic", topic,
+			"message_id", msg.ID,
+			"error", err,
+		)
+		return true
+	}
+	if !entry.Blocked() {
+		return true
+	}
+
+	// The delivery is blocked; only now is the parked descriptor built and
+	// the queue mechanics arranged. The wait context is cancellable by the
+	// keep-in-flight timer: when a visibility extension fails, the delivery
+	// can no longer be held safely (the queue may already be re-delivering
+	// it), so the wait is cancelled with the extension failure as the cause.
+	parked := consumergate.Parked{
+		Topic:     topic,
+		MessageID: msg.ID,
+		Payload:   msg.Payload,
+		Attempt:   delivery.Attempt(),
+	}
+	waitCtx, cancelWait := context.WithCancelCause(ctx)
+	defer cancelWait(nil)
+
+	stopExtending := m.keepInFlight(waitCtx, delivery, cancelWait)
+	start := time.Now()
+	err = entry.Wait(waitCtx, parked)
+	stopExtending()
+	metrics.NamedLatencyHistogram(scope, opName, "wait_latency", time.Since(start))
+
+	if err == nil {
+		return true
+	}
+
+	if errors.Is(err, context.Canceled) {
+		// Two cancellation sources share the context error; the cancellation
+		// cause tells them apart. A keep-in-flight failure carries itself as
+		// the cause; a consumer shutdown propagates the plain context.Canceled.
+		if cause := context.Cause(waitCtx); cause != nil && !errors.Is(cause, context.Canceled) {
+			metrics.NamedCounter(scope, opName, "extend_errors", 1)
+			m.logger.Errorw("failed to keep gate-blocked delivery in-flight, failing open",
+				"consumer_group", consumerGroup,
+				"topic", topic,
+				"message_id", msg.ID,
+				"error", cause,
+			)
+			return true
+		}
+		metrics.NamedCounter(scope, opName, "shutdown_while_blocked", 1)
+		m.logger.Infow("consumer shutdown while delivery was blocked by gate",
+			"consumer_group", consumerGroup,
+			"topic", topic,
+			"message_id", msg.ID,
+		)
+		return false
+	}
+
+	// Gate state could not be re-read or the release could not be recorded:
+	// fail open, as above.
+	metrics.NamedCounter(scope, opName, "wait_errors", 1)
+	m.logger.Errorw("gate wait failed, failing open",
+		"consumer_group", consumerGroup,
+		"topic", topic,
+		"message_id", msg.ID,
+		"error", err,
+	)
+	return true
+}
+
+// keepInFlight periodically extends a delivery's visibility while it is
+// blocked behind its gate, so the queue does not redeliver a held message.
+// It is timer-driven rather than goroutine-driven: nothing runs unless the
+// delivery actually stays blocked past the first extend interval, so the
+// common case — an open gate — costs one timer arm/disarm and no goroutine.
+//
+// An extension failure means the delivery can no longer be held safely (its
+// visibility may lapse and the queue redeliver it), so the failure propagates
+// through cancelWait — unblocking the gate wait immediately — and the timer
+// chain stops. The returned stop function disarms the timer and waits out any
+// in-flight extension, guaranteeing no extension races the controller.
+func (m *consumer) keepInFlight(ctx context.Context, delivery extqueue.Delivery, cancelWait context.CancelCauseFunc) (stop func()) {
+	var mu sync.Mutex
+	stopped := false
+
+	// The timer variable is assigned under the lock so the callback — which
+	// may fire before AfterFunc returns — observes the assignment through the
+	// same lock it takes before Reset.
+	var timer *time.Timer
+	mu.Lock()
+	timer = time.AfterFunc(m.gateExtendInterval, func() {
+		// The lock is held across the extension so stop() joins an in-flight
+		// callback before returning.
+		mu.Lock()
+		defer mu.Unlock()
+		if stopped {
+			return
+		}
+		if err := delivery.ExtendVisibilityTimeout(ctx, gateExtensionMs); err != nil {
+			cancelWait(fmt.Errorf("failed to extend visibility of gate-blocked delivery: %w", err))
+			return
+		}
+		timer.Reset(m.gateExtendInterval)
+	})
+	mu.Unlock()
+
+	return func() {
+		mu.Lock()
+		stopped = true
+		mu.Unlock()
+		timer.Stop()
+	}
+}
+
 // Stop gracefully shuts down all handlers with the specified timeout.
 // Cancels all subscription contexts and waits for consumption goroutines to finish.
 // timeoutMs is the maximum time in milliseconds to wait for graceful shutdown.