feat(consumergate): runtime stop/start of queue controllers

Implement the consumer-gate RFC with a file-backed gate shared by gateway,
orchestrator, and runway consumers.

- clear deliveries through a consumer-group/partition gate before controller
  processing while extending visibility for blocked deliveries
- expose caller-owned delivery descriptors and let gate implementations stamp
  gate-owned parked-record fields
- keep parked payload files only while deliveries are actively blocked and
  remove them on every terminal watch path
- fail open on gate or visibility-extension failures and leave blocked
  deliveries unacked during shutdown for normal redelivery
- run service containers with the host test UID/GID under rootful Docker while
  preserving rootless Docker ownership mapping
- make the cancellation E2E scenario deterministic by parking the runway
  merge-conflict-check delivery until cancellation reaches a terminal state
- consolidate consumer white-box and behavioral unit tests in the consumer
  package

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/validator/... ./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/validator/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
 	@echo "Mocks generated successfully!"
 
 proto: ## Generate protobuf files from .proto definitions

doc/rfc/consumer-gate.md

@@ -29,7 +29,7 @@ Every controller subscribes with a unique consumer group (`orchestrator-batch`,
 
 ### Gate state is a separate extension
 
-The consumer gate is a shared extension in its own right, not a feature of any queue backend. The contract lives at `platform/extension/consumergate/`: the behavioral interface the middleware reads (is this group/partition gated? record a parked delivery, record its release), the write surface tests and tooling use (close a gate, open it), and the `Config`. Implementations live in subdirectories, per the standard extension layout. The consumer package takes the read-side interface as a dependency — wiring constructs an implementation and passes it to `consumer.New` via a new option; when no gate is configured, the middleware is absent and the consumer behaves exactly as today. The wiring delta is one option argument at each consumer construction site (gateway, orchestrator primary, orchestrator DLQ, runway); no per-controller wiring, and DLQ consumers are gated uniformly with the rest.
+The consumer gate is a shared extension in its own right, not a feature of any queue backend. The contract lives at `platform/extension/consumergate/`: the behavioral interface the middleware reads (is this group/partition gated? record a parked delivery, record its release), the write surface tests and tooling use (close a gate, open it), and the `Config`. `Watch` accepts a caller-owned `DeliveryDescriptor` containing only message data; the implementation combines it with the gate identity captured by `Enter` and its own timestamp to create the observable `Parked` record, so callers cannot supply or overwrite gate-owned fields. Implementations live in subdirectories, per the standard extension layout. The consumer package takes the read-side interface as a dependency — wiring constructs an implementation and passes it to `consumer.New` via a new option; when no gate is configured, the middleware is absent and the consumer behaves exactly as today. The wiring delta is one option argument at each consumer construction site (gateway, orchestrator primary, orchestrator DLQ, runway); no per-controller wiring, and DLQ consumers are gated uniformly with the rest.
 
 Keeping the contract separate from any backend is what lets the storage medium be chosen per deployment: a filesystem directory first (below), a database- or config-service-backed implementation later if fleet-wide coordination demands it — with the middleware, the wiring shape, and every test written against the contract unchanged.
 
@@ -43,25 +43,25 @@ The first implementation stores gate state as plain files under a configured dir
 {dir}/parked/{consumer_group}/{topic}/{urlenc(id)}.json  # one parked delivery record
 ```
 
-Consumer groups and topics are already filesystem-safe by the repo's naming rules; partition keys and message IDs may contain `/` (request IDs like `queue/1`), so they are URL-encoded in file names. Gate files contain human-readable JSON metadata — `reason`, `created_by`, `created_at_ms` — so an operator finding a paused controller can tell why. Parked records carry the payload, attempt, `parked_at_ms`, and a `released_at_ms` stamped when the delivery proceeds; all writes go through temp-file-plus-rename so readers never see partial JSON.
+Consumer groups and topics are already filesystem-safe by the repo's naming rules; partition keys and message IDs may contain `/` (request IDs like `queue/1`), so they are URL-encoded in file names. Gate files contain human-readable JSON metadata — `reason`, `created_by`, `created_at_ms` — so an operator finding a paused controller can tell why. Parked records carry the payload, attempt, and `parked_at_ms` while a delivery is blocked; the record is deleted before the wait ends, so payloads are not retained after release, cancellation, or monitoring failure. All writes go through temp-file-plus-rename so readers never see partial JSON.
 
 Files are the simplest medium that satisfies every requirement in this RFC, and simplicity is the point of the first implementation:
 
 - **Operator interface for free.** Pausing a controller is writing a small file; resuming is `rm`. Inspecting a paused stage is `ls` and `cat`. No client, no schema, no query.
 - **Trivially reachable out of process.** In the e2e stack, the compose file bind-mounts a host directory into every service container at a fixed path (passed via one environment variable); the test process manipulates gates and reads parked records as local files. In single-host dev the same directory works as-is.
 - **Durable and independent.** State survives service restarts — a paused stage stays paused until explicitly opened — and the gate has no dependency on the queue backend or any database being healthy.
 
-The middleware **polls** the directory rather than using filesystem notifications: inotify events do not propagate reliably across bind mounts and overlay filesystems, and the cached-poll posture (below) makes notification latency irrelevant. The known limit of the file medium is multi-replica fleets: a file gates the replicas that see the directory, so a fleet-wide pause needs the deployment platform to distribute the file — or a future store-backed implementation of the same contract. That trade is accepted; the deployments this RFC serves (e2e, single-host dev, per-instance operational pause) are exactly where files excel.
+The middleware **polls** the directory rather than using filesystem notifications: inotify is platform-specific, watches can overflow or require re-registration, and event behavior varies across bind mounts, overlay or network filesystems, rootless Docker, and Docker Desktop's host/container filesystem bridge. Polling is the portable convergence mechanism; filesystem events may be added later as an optional wakeup optimization alongside it. The known limit of the file medium is multi-replica fleets: a file gates the replicas that see the directory, so a fleet-wide pause needs the deployment platform to distribute the file — or a future store-backed implementation of the same contract. That trade is accepted; the deployments this RFC serves (e2e, single-host dev, per-instance operational pause) are exactly where files excel.
 
-### Read path: cached poll, bounded effect latency
+### Read path: direct reads and bounded release latency
 
-The middleware does not check gate state per message. Gate state is cached per controller and refreshed on a short interval (configurable, ~1s), and a parked delivery re-checks on the same tick. The dormant cost of the feature — the common case, forever — is one directory stat per controller per interval. The price is that closing a gate takes effect within one refresh interval plus the in-flight message's completion; opening one takes effect within one interval.
+The middleware checks the applicable gate files for every delivery. A parked delivery re-checks them on a short interval (configurable, ~1s). Closing a gate therefore affects the next delivery check without waiting for a cache refresh; opening one releases already parked deliveries within one poll interval.
 
 Tests do not depend on that latency. The deterministic patterns are two: **arrange first** (close the gate before publishing the message that must be caught — exact by construction), or **await the observed effect** (the parked record, below) instead of assuming timing.
 
 ### Observation: parked deliveries are recorded
 
-Parking writes the parked record before blocking. This record is the "observe" half of stop/observe/start: a test awaits the record to *know* the stop caught its message (there is otherwise no signal distinguishing "gated and parked" from "not arrived yet"), can assert on the recorded payload, and can decide what to do next while the controller is provably stopped. For an operator, the same records answer "what is this paused controller holding?". Records are bounded by parked messages, which are bounded by gate usage; the directory is empty whenever no gate is in use.
+Parking writes the parked record before blocking. This record is the "observe" half of stop/observe/start: a test awaits the record to *know* the stop caught its message (there is otherwise no signal distinguishing "gated and parked" from "not arrived yet"), can assert on the recorded payload, and can decide what to do next while the controller is provably stopped. For an operator, the same records answer "what is this paused controller holding?". The record is removed before the wait reports release, cancellation, or failure, so records are bounded by currently parked messages and the directory is empty whenever no delivery is held behind a gate.
 
 ### Failure posture: fail open
 

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",
@@ -25,11 +26,12 @@ go_test(
         "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,128 @@ 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. For a blocked entry the gate hands back a
+// watch channel (its own monitoring goroutine behind it), and this routine
+// multiplexes three events in a single select loop — the watch channel, a
+// visibility-extension ticker that keeps the blocked delivery in-flight, and
+// parent-context cancellation — with no extra goroutine on the consumer side.
+// The watch context is a child of ctx cancelled on every return path, so the
+// gate's goroutine always exits and nothing is left dangling. 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 {
+		if errors.Is(err, context.Canceled) {
+			// Cancellation is in progress; return false per the contract.
+			return false
+		}
+		// 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
+	}
+
+	start := time.Now()
+	defer func() {
+		metrics.NamedHistogram(scope, opName, "wait_latency", metrics.DefaultLatencyBuckets).RecordDuration(time.Since(start))
+	}()
+
+	// The delivery is blocked. Build the caller-owned descriptor and start
+	// watching the gate; watchCtx is cancelled on every return path so the gate's
+	// monitoring goroutine exits even when we fail open with the consumer still
+	// running.
+	descriptor := consumergate.DeliveryDescriptor{
+		Topic:     topic,
+		MessageID: msg.ID,
+		Payload:   msg.Payload,
+		Attempt:   delivery.Attempt(),
+	}
+
+	// Start a child context for the watch and cancel it on every return path,
+	// so the gate's monitoring goroutine always exits.
+	watchCtx, cancelWatch := context.WithCancel(ctx)
+	defer cancelWatch()
+	watchCh := entry.Watch(watchCtx, descriptor)
+
+	// Keep the blocked delivery in-flight on a ticker while multiplexing the
+	// gate watch and parent cancellation.
+	ticker := time.NewTicker(m.gateExtendInterval)
+	defer ticker.Stop()
+
+loop:
+	for {
+		select {
+		case e := <-watchCh:
+			if e == nil {
+				// The gate opened; the delivery proceeds. Returning is safe:
+				// the deferred cancelWatch stops the gate's monitoring
+				// goroutine and the deferred ticker.Stop halts the extender.
+				return true
+			}
+			if err == nil {
+				// This includes a cancellation error propagated from the
+				// parent context.
+				err = e
+			}
+			break loop
+
+		case <-ticker.C:
+			if err != nil {
+				// The error is already set, so this tick is bogus and can be
+				// skipped; wait for the gate watch to report back and end the
+				// loop.
+				continue
+			}
+			if e := delivery.ExtendVisibilityTimeout(ctx, gateExtensionMs); e != nil {
+				// The delivery can no longer be held safely, so cancel the gate
+				// watch; it reports back on watchCh and ends the loop.
+				err = e
+				cancelWatch()
+			}
+		}
+	}
+
+	// The loop only breaks with a non-nil err.
+	if errors.Is(err, context.Canceled) {
+		// Cancellation is in progress; return false per the contract.
+		return false
+	}
+	// Gate state could not be re-read or the record could not be
+	// written: 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
+}
+
 // 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.