From 617595bbb33b07fc52f8bb1336db4d6773a587f8 Mon Sep 17 00:00:00 2001
From: sergeyb <sergeyb@uber.com>
Date: Wed, 15 Jul 2026 18:21:11 +0000
Subject: [PATCH] feat(consumergate): runtime stop/start of queue controllers +
 deterministic e2e cancel test
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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>
---
 Makefile                                      |   2 +-
 platform/consumer/BUILD.bazel                 |   6 +-
 platform/consumer/consumer.go                 | 207 ++++++++-
 platform/consumer/consumer_internal_test.go   | 179 ++++++++
 platform/consumer/consumer_test.go            | 311 ++++++++++++-
 platform/extension/consumergate/BUILD.bazel   |   8 +
 platform/extension/consumergate/README.md     |  23 +
 .../extension/consumergate/consumergate.go    | 163 +++++++
 .../extension/consumergate/file/BUILD.bazel   |  20 +
 .../extension/consumergate/file/README.md     |  21 +
 platform/extension/consumergate/file/store.go | 417 ++++++++++++++++++
 .../extension/consumergate/file/store_test.go | 326 ++++++++++++++
 .../extension/consumergate/mock/BUILD.bazel   |  12 +
 .../consumergate/mock/consumergate_mock.go    | 215 +++++++++
 .../extension/consumergate/noop/BUILD.bazel   |  20 +
 .../extension/consumergate/noop/README.md     |   3 +
 platform/extension/consumergate/noop/gate.go  |  49 ++
 .../extension/consumergate/noop/gate_test.go  |  37 ++
 platform/metrics/metrics.go                   |   7 +
 service/runway/server/BUILD.bazel             |   2 +
 service/runway/server/main.go                 |  23 +
 service/stovepipe/server/BUILD.bazel          |   1 +
 service/stovepipe/server/main.go              |   6 +-
 service/submitqueue/docker-compose.yml        |  18 +
 .../submitqueue/gateway/server/BUILD.bazel    |   2 +
 service/submitqueue/gateway/server/main.go    |  23 +
 .../orchestrator/server/BUILD.bazel           |   2 +
 .../submitqueue/orchestrator/server/main.go   |  21 +
 .../extension/storage/mock/storage_mock.go    |  28 +-
 test/e2e/submitqueue/BUILD.bazel              |   7 +
 test/e2e/submitqueue/harness_test.go          |  76 ++++
 test/e2e/submitqueue/suite_test.go            | 109 ++++-
 .../submitqueue/core/consumer/BUILD.bazel     |   1 +
 .../core/consumer/consumer_test.go            |   3 +-
 34 files changed, 2284 insertions(+), 64 deletions(-)
 create mode 100644 platform/consumer/consumer_internal_test.go
 create mode 100644 platform/extension/consumergate/BUILD.bazel
 create mode 100644 platform/extension/consumergate/README.md
 create mode 100644 platform/extension/consumergate/consumergate.go
 create mode 100644 platform/extension/consumergate/file/BUILD.bazel
 create mode 100644 platform/extension/consumergate/file/README.md
 create mode 100644 platform/extension/consumergate/file/store.go
 create mode 100644 platform/extension/consumergate/file/store_test.go
 create mode 100644 platform/extension/consumergate/mock/BUILD.bazel
 create mode 100644 platform/extension/consumergate/mock/consumergate_mock.go
 create mode 100644 platform/extension/consumergate/noop/BUILD.bazel
 create mode 100644 platform/extension/consumergate/noop/README.md
 create mode 100644 platform/extension/consumergate/noop/gate.go
 create mode 100644 platform/extension/consumergate/noop/gate_test.go

diff --git a/Makefile b/Makefile
index d80bd17e..831ed58b 100644
--- a/Makefile
+++ b/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
diff --git a/platform/consumer/BUILD.bazel b/platform/consumer/BUILD.bazel
index c0e54aa7..80b20930 100644
--- a/platform/consumer/BUILD.bazel
+++ b/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",
diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go
index 0569ba2d..f42d003f 100644
--- a/platform/consumer/consumer.go
+++ b/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,169 @@ 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: the entry wait and a keep-in-flight extender
+// run as goroutines under their own child contexts, supervised by this
+// routine, which cancels and joins both before the delivery proceeds or is
+// dropped. 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 are the parked descriptor built and
+	// the queue mechanics arranged. The main routine supervises two
+	// goroutines — the entry wait and the keep-in-flight extender — each
+	// under its own child context, and reacts to whichever finishes first.
+	parked := consumergate.Parked{
+		Topic:     topic,
+		MessageID: msg.ID,
+		Payload:   msg.Payload,
+		Attempt:   delivery.Attempt(),
+	}
+	waitCtx, cancelWait := context.WithCancel(ctx)
+	defer cancelWait()
+	extendCtx, cancelExtend := context.WithCancel(ctx)
+
+	extendErrCh, extendDone := m.keepInFlight(extendCtx, delivery)
+	// Whatever path exits, cancel the extender and join it, so no extension
+	// races the controller (or the redelivery, when the delivery is dropped).
+	defer func() {
+		cancelExtend()
+		<-extendDone
+	}()
+
+	start := time.Now()
+	defer func() {
+		metrics.NamedLatencyHistogram(scope, opName, "wait_latency", time.Since(start))
+	}()
+	waitResult := make(chan error, 1)
+	go func() {
+		waitResult <- entry.Wait(waitCtx, parked)
+	}()
+
+	select {
+	case err = <-waitResult:
+		// The wait finished on its own; fall through to classify its result.
+	case extendErr := <-extendErrCh:
+		// The delivery can no longer be held safely (its visibility may lapse
+		// and the queue redeliver it): cancel the wait and join it, then fail
+		// open — unless the consumer is shutting down anyway, in which case
+		// the delivery is dropped for redelivery like any other shutdown.
+		cancelWait()
+		<-waitResult
+		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", consumerGroup,
+				"topic", topic,
+				"message_id", msg.ID,
+			)
+			return false
+		}
+		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", extendErr,
+		)
+		return true
+	}
+
+	if err == nil {
+		return true
+	}
+
+	if errors.Is(err, context.Canceled) {
+		// On this path the wait context was never cancelled by the extend
+		// branch above, so a cancelled wait means the parent was cancelled:
+		// the consumer is shutting down and the delivery is dropped without
+		// processing or ack/nack.
+		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 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
+}
+
+// keepInFlight starts a goroutine that periodically extends the delivery's
+// visibility so the queue does not redeliver a gate-blocked message. The
+// goroutine's lifecycle is controlled by ctx: it extends on each tick until
+// ctx is cancelled or an extension fails. A failure means the delivery can no
+// longer be held safely (its visibility may lapse and the queue redeliver
+// it); it is sent on the returned error channel for the supervising routine
+// to act on, and ends the goroutine.
+//
+// The returned done channel closes when the goroutine exits; callers cancel
+// ctx and receive from done to join the goroutine before letting the delivery
+// proceed, guaranteeing no extension races the controller.
+func (m *consumer) keepInFlight(ctx context.Context, delivery extqueue.Delivery) (extendErr <-chan error, done <-chan struct{}) {
+	errCh := make(chan error, 1)
+	doneCh := make(chan struct{})
+
+	go func() {
+		defer close(doneCh)
+		ticker := time.NewTicker(m.gateExtendInterval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
+			if err := delivery.ExtendVisibilityTimeout(ctx, gateExtensionMs); err != nil {
+				errCh <- fmt.Errorf("failed to extend visibility of gate-blocked delivery: %w", err)
+				return
+			}
+		}
+	}()
+
+	return errCh, doneCh
+}
+
 // 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.
diff --git a/platform/consumer/consumer_internal_test.go b/platform/consumer/consumer_internal_test.go
new file mode 100644
index 00000000..a5ee8a71
--- /dev/null
+++ b/platform/consumer/consumer_internal_test.go
@@ -0,0 +1,179 @@
+// 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
+
+// In-package tests for the gate keep-in-flight path, which needs a short
+// extend interval (the gateExtendInterval field) to be exercised quickly.
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/uber-go/tally"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	"github.com/uber/submitqueue/platform/errs"
+	"github.com/uber/submitqueue/platform/extension/consumergate"
+	consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop"
+	"go.uber.org/zap/zaptest"
+)
+
+// fakeExtendDelivery is a hand-rolled fake delivery whose visibility
+// extensions are observable on a channel and can be forced to fail.
+type fakeExtendDelivery struct {
+	msg       entityqueue.Message
+	extendErr error      // returned by every ExtendVisibilityTimeout call
+	extended  chan int64 // receives the requested extension on each call
+}
+
+func (d *fakeExtendDelivery) Message() entityqueue.Message             { return d.msg }
+func (d *fakeExtendDelivery) Attempt() int                             { return 1 }
+func (d *fakeExtendDelivery) ReceivedAt() int64                        { return 0 }
+func (d *fakeExtendDelivery) Metadata() map[string]string              { return nil }
+func (d *fakeExtendDelivery) DeliveryID() string                       { return d.msg.ID }
+func (d *fakeExtendDelivery) Ack(_ context.Context) error              { return nil }
+func (d *fakeExtendDelivery) Nack(_ context.Context, _ int64) error    { return nil }
+func (d *fakeExtendDelivery) Reject(_ context.Context, _ string) error { return nil }
+func (d *fakeExtendDelivery) ExtendVisibilityTimeout(_ context.Context, ms int64) error {
+	d.extended <- ms
+	return d.extendErr
+}
+
+// newGateTestConsumer builds a consumer with a fast keep-in-flight interval.
+func newGateTestConsumer(t *testing.T, gate consumergate.Gate) *consumer {
+	t.Helper()
+	reg, err := NewTopicRegistry(nil)
+	require.NoError(t, err)
+	c := New(zaptest.NewLogger(t).Sugar(), tally.NoopScope, reg, errs.NewClassifierProcessor(), gate).(*consumer)
+	c.gateExtendInterval = 5 * time.Millisecond
+	return c
+}
+
+func newFakeExtendDelivery(extendErr error) *fakeExtendDelivery {
+	return &fakeExtendDelivery{
+		msg:       entityqueue.NewMessage("msg-1", []byte("p"), "part", nil),
+		extendErr: extendErr,
+		extended:  make(chan int64, 16),
+	}
+}
+
+func TestKeepInFlight_ExtendsAndJoinsOnCancel(t *testing.T) {
+	c := newGateTestConsumer(t, consumergatenoop.New())
+	del := newFakeExtendDelivery(nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	extendErr, done := c.keepInFlight(ctx, del)
+
+	// The extender ticks and extends with the gate extension budget.
+	select {
+	case ms := <-del.extended:
+		assert.Equal(t, gateExtensionMs, ms)
+	case <-time.After(200 * time.Millisecond):
+		t.Fatal("timed out waiting for ExtendVisibilityTimeout call")
+	}
+
+	// Cancelling the context ends the goroutine; done closing is the join.
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(200 * time.Millisecond):
+		t.Fatal("timed out waiting for keepInFlight goroutine to exit")
+	}
+	select {
+	case err := <-extendErr:
+		t.Fatalf("successful extensions must not signal an error, got %v", err)
+	default:
+	}
+}
+
+func TestKeepInFlight_ExtendFailureSignalsAndExits(t *testing.T) {
+	c := newGateTestConsumer(t, consumergatenoop.New())
+	failure := fmt.Errorf("visibility store unavailable")
+	del := newFakeExtendDelivery(failure)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	extendErr, done := c.keepInFlight(ctx, del)
+
+	// The failed extension must be signalled to the supervising routine and
+	// end the goroutine.
+	select {
+	case err := <-extendErr:
+		assert.ErrorIs(t, err, failure)
+	case <-time.After(200 * time.Millisecond):
+		t.Fatal("timed out waiting for extend failure signal")
+	}
+	select {
+	case <-done:
+	case <-time.After(200 * time.Millisecond):
+		t.Fatal("timed out waiting for keepInFlight goroutine to exit")
+	}
+}
+
+// blockedGate always parks: Enter returns a blocked entry whose Wait blocks
+// until its context is cancelled, mimicking a closed gate that never opens.
+type blockedGate struct{}
+
+func (blockedGate) Enter(context.Context, consumergate.Key) (consumergate.Entry, error) {
+	return blockedEntry{}, nil
+}
+
+// blockedEntry is blockedGate's never-opening entry.
+type blockedEntry struct{}
+
+func (blockedEntry) Blocked() bool { return true }
+
+func (blockedEntry) Wait(ctx context.Context, _ consumergate.Parked) error {
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+// fakeGateController is a minimal Controller for driving waitGate directly.
+type fakeGateController struct{}
+
+func (fakeGateController) Process(context.Context, Delivery) error { return nil }
+func (fakeGateController) Name() string                            { return "fake" }
+func (fakeGateController) TopicKey() TopicKey                      { return TopicKey("topic") }
+func (fakeGateController) ConsumerGroup() string                   { return "group" }
+
+func TestWaitGate_ExtendFailureFailsOpen(t *testing.T) {
+	c := newGateTestConsumer(t, blockedGate{})
+	del := newFakeExtendDelivery(fmt.Errorf("visibility store unavailable"))
+
+	// The gate never opens, so only the extension failure can end the wait —
+	// and it must end it by failing open, not by dropping the delivery.
+	proceed := c.waitGate(context.Background(), fakeGateController{}, del, tally.NoopScope)
+	assert.True(t, proceed, "a delivery that can no longer be held must fail open")
+}
+
+func TestWaitGate_ShutdownWhileBlockedDrops(t *testing.T) {
+	c := newGateTestConsumer(t, blockedGate{})
+	del := newFakeExtendDelivery(nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan bool, 1)
+	go func() { done <- c.waitGate(ctx, fakeGateController{}, del, tally.NoopScope) }()
+	cancel()
+
+	select {
+	case proceed := <-done:
+		assert.False(t, proceed, "consumer shutdown must drop the blocked delivery for redelivery")
+	case <-time.After(200 * time.Millisecond):
+		t.Fatal("timed out waiting for waitGate to observe shutdown")
+	}
+}
diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go
index 1bf4682f..1ad8579f 100644
--- a/platform/consumer/consumer_test.go
+++ b/platform/consumer/consumer_test.go
@@ -30,6 +30,8 @@ import (
 	"github.com/uber/submitqueue/platform/consumer"
 	consumermock "github.com/uber/submitqueue/platform/consumer/mock"
 	"github.com/uber/submitqueue/platform/errs"
+	"github.com/uber/submitqueue/platform/extension/consumergate"
+	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 +94,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 +103,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 +123,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 +143,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 +159,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 +170,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 +196,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 +222,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 +248,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 +294,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 +336,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 +387,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 +445,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 +520,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 +565,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 +621,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 +706,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 +775,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)
 
@@ -825,7 +827,7 @@ func TestConsumer_ConsumeLoopSurvivesCallerDeadline(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())
 
 	processed := make(chan string, 1)
 	handler := consumermock.NewMockController(ctrl)
@@ -860,3 +862,274 @@ func TestConsumer_ConsumeLoopSurvivesCallerDeadline(t *testing.T) {
 	err = c.Stop(30000)
 	require.NoError(t, err)
 }
+
+// fakeGate is a channel-instrumented consumergate.Gate so tests can await the
+// park/release transitions instead of sleeping.
+type fakeGate struct {
+	mu     sync.Mutex
+	closed map[consumergate.Key]bool
+	err    error
+
+	parked   chan consumergate.Parked
+	released chan string // message IDs
+}
+
+func newFakeGate() *fakeGate {
+	return &fakeGate{
+		closed:   make(map[consumergate.Key]bool),
+		parked:   make(chan consumergate.Parked, 16),
+		released: make(chan string, 16),
+	}
+}
+
+func (f *fakeGate) setClosed(key consumergate.Key, closed bool) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.closed[key] = closed
+}
+
+func (f *fakeGate) setErr(err error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.err = err
+}
+
+func (f *fakeGate) isClosed(consumerGroup, partitionKey string) bool {
+	if f.closed[consumergate.Key{ConsumerGroup: consumerGroup}] {
+		return true
+	}
+	return f.closed[consumergate.Key{ConsumerGroup: consumerGroup, PartitionKey: partitionKey}]
+}
+
+// Enter implements consumergate.Gate. It checks the err field first, then
+// returns an unblocked entry for an open gate or a blocked entry for a closed
+// one. The blocked entry's Wait mimics the contract: it stamps the entered
+// identity on the parked descriptor, announces it on the parked channel, and
+// polls on a 2ms timer until the gate opens or ctx is cancelled; on open the
+// entry sends the message ID on the released channel.
+func (f *fakeGate) Enter(_ context.Context, key consumergate.Key) (consumergate.Entry, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.err != nil {
+		return nil, f.err
+	}
+	if !f.isClosed(key.ConsumerGroup, key.PartitionKey) {
+		return fakeOpenEntry{}, nil
+	}
+	return &fakeBlockedEntry{gate: f, key: key}, nil
+}
+
+// fakeOpenEntry is the entry handed out for an open fake gate.
+type fakeOpenEntry struct{}
+
+func (fakeOpenEntry) Blocked() bool { return false }
+
+func (fakeOpenEntry) Wait(context.Context, consumergate.Parked) error { return nil }
+
+// fakeBlockedEntry is the entry handed out for a closed fake gate.
+type fakeBlockedEntry struct {
+	gate *fakeGate
+	key  consumergate.Key
+}
+
+func (*fakeBlockedEntry) Blocked() bool { return true }
+
+func (e *fakeBlockedEntry) Wait(ctx context.Context, parked consumergate.Parked) error {
+	parked.ConsumerGroup = e.key.ConsumerGroup
+	parked.PartitionKey = e.key.PartitionKey
+	e.gate.parked <- parked
+
+	ticker := time.NewTicker(2 * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+			e.gate.mu.Lock()
+			closed := e.gate.isClosed(e.key.ConsumerGroup, e.key.PartitionKey)
+			e.gate.mu.Unlock()
+			if !closed {
+				e.gate.released <- parked.MessageID
+				return nil
+			}
+		}
+	}
+}
+
+// startGatedConsumer builds a consumer with the fake gate directly as the 5th
+// New arg, one registered mock controller, and a live subscription fed by the
+// returned delivery channel.
+func startGatedConsumer(t *testing.T, ctrl *gomock.Controller, gate consumergate.Gate, processFunc func(context.Context, consumer.Delivery) error) (consumer.Consumer, chan extqueue.Delivery) {
+	t.Helper()
+
+	deliveryChan := make(chan extqueue.Delivery, 4)
+	mockSub := queuemock.NewMockSubscriber(ctrl)
+	mockSub.EXPECT().Subscribe(gomock.Any(), gomock.Any(), gomock.Any()).Return(deliveryChan, nil)
+
+	mockQ := queuemock.NewMockQueue(ctrl)
+	mockQ.EXPECT().Subscriber().Return(mockSub)
+
+	reg := newRegistry(t, mockQ, topickey.TopicKeyStart, "test-group")
+
+	c := consumer.New(zaptest.NewLogger(t).Sugar(), tally.NoopScope, reg, errs.NewClassifierProcessor(), gate)
+
+	handler := consumermock.NewMockController(ctrl)
+	setupController(handler, "test-handler", topickey.TopicKeyStart, "test-group", processFunc)
+	require.NoError(t, c.Register(handler))
+
+	require.NoError(t, c.Start(context.Background()))
+	return c, deliveryChan
+}
+
+// gatedDelivery builds a MockDelivery that also tolerates visibility
+// extensions while parked.
+func gatedDelivery(ctrl *gomock.Controller, msg entityqueue.Message) (*queuemock.MockDelivery, chan struct{}) {
+	mockDel := queuemock.NewMockDelivery(ctrl)
+	done := setupDelivery(mockDel, msg, nil, nil)
+	mockDel.EXPECT().ExtendVisibilityTimeout(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	return mockDel, done
+}
+
+func TestConsumer_Gate_OpenGatePassesThrough(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	gate := newFakeGate()
+
+	handledMsg := ""
+	c, deliveryChan := startGatedConsumer(t, ctrl, gate, func(_ context.Context, delivery consumer.Delivery) error {
+		handledMsg = delivery.Message().ID
+		return nil
+	})
+
+	msg := entityqueue.NewMessage("msg-1", []byte("payload"), "partition1", nil)
+	mockDel, done := gatedDelivery(ctrl, msg)
+
+	deliveryChan <- mockDel
+	<-done
+
+	assert.Equal(t, "msg-1", handledMsg)
+	assert.Empty(t, gate.parked, "an open gate must not park deliveries")
+
+	require.NoError(t, c.Stop(30000))
+}
+
+func TestConsumer_Gate_ParksThenReleases(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	gate := newFakeGate()
+	gate.setClosed(consumergate.Key{ConsumerGroup: "test-group"}, true)
+
+	var processed atomic.Bool
+	c, deliveryChan := startGatedConsumer(t, ctrl, gate, func(context.Context, consumer.Delivery) error {
+		processed.Store(true)
+		return nil
+	})
+
+	msg := entityqueue.NewMessage("msg-1", []byte("payload"), "partition1", nil)
+	mockDel, done := gatedDelivery(ctrl, msg)
+	deliveryChan <- mockDel
+
+	// The parked record is written before the gate blocks, so awaiting it
+	// proves the gate caught the message before the controller saw it.
+	parked := <-gate.parked
+	assert.Equal(t, "test-group", parked.ConsumerGroup)
+	assert.Equal(t, topickey.TopicKeyStart.String(), parked.Topic)
+	assert.Equal(t, "msg-1", parked.MessageID)
+	assert.Equal(t, "partition1", parked.PartitionKey)
+	assert.Equal(t, []byte("payload"), parked.Payload)
+	assert.Equal(t, 1, parked.Attempt)
+	assert.False(t, processed.Load(), "controller must not run while its gate is closed")
+
+	// Open the gate: the parked delivery proceeds, the release is recorded,
+	// and the message is acked.
+	gate.setClosed(consumergate.Key{ConsumerGroup: "test-group"}, false)
+	assert.Equal(t, "msg-1", <-gate.released)
+	<-done
+	assert.True(t, processed.Load())
+
+	require.NoError(t, c.Stop(30000))
+}
+
+func TestConsumer_Gate_PartitionScoped(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	gate := newFakeGate()
+	gate.setClosed(consumergate.Key{ConsumerGroup: "test-group", PartitionKey: "gated-partition"}, true)
+
+	var handled sync.Map
+	c, deliveryChan := startGatedConsumer(t, ctrl, gate, func(_ context.Context, delivery consumer.Delivery) error {
+		handled.Store(delivery.Message().ID, true)
+		return nil
+	})
+
+	gatedMsg := entityqueue.NewMessage("gated-msg", []byte("p"), "gated-partition", nil)
+	gatedDel, gatedDone := gatedDelivery(ctrl, gatedMsg)
+	openMsg := entityqueue.NewMessage("open-msg", []byte("p"), "open-partition", nil)
+	openDel, openDone := gatedDelivery(ctrl, openMsg)
+
+	deliveryChan <- gatedDel
+	parked := <-gate.parked
+	assert.Equal(t, "gated-msg", parked.MessageID)
+
+	// Unrelated traffic keeps flowing through the same controller while one
+	// partition is parked.
+	deliveryChan <- openDel
+	<-openDone
+	_, ok := handled.Load("open-msg")
+	assert.True(t, ok)
+	_, ok = handled.Load("gated-msg")
+	assert.False(t, ok)
+
+	gate.setClosed(consumergate.Key{ConsumerGroup: "test-group", PartitionKey: "gated-partition"}, false)
+	<-gatedDone
+	_, ok = handled.Load("gated-msg")
+	assert.True(t, ok)
+
+	require.NoError(t, c.Stop(30000))
+}
+
+func TestConsumer_Gate_ShutdownWhileParked(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	gate := newFakeGate()
+	gate.setClosed(consumergate.Key{ConsumerGroup: "test-group"}, true)
+
+	var processed atomic.Bool
+	c, deliveryChan := startGatedConsumer(t, ctrl, gate, func(context.Context, consumer.Delivery) error {
+		processed.Store(true)
+		return nil
+	})
+
+	msg := entityqueue.NewMessage("msg-1", []byte("payload"), "partition1", nil)
+	mockDel, _ := gatedDelivery(ctrl, msg)
+	deliveryChan <- mockDel
+	<-gate.parked
+
+	// Stopping while parked must not stall shutdown, must not invoke the
+	// controller, and must not ack/nack — the delivery is left in-flight for
+	// redelivery after its visibility lapses.
+	require.NoError(t, c.Stop(30000))
+	assert.False(t, processed.Load())
+	assert.Empty(t, gate.released, "a delivery dropped at shutdown is not released")
+}
+
+func TestConsumer_Gate_FailsOpenOnReadError(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	gate := newFakeGate()
+	gate.setClosed(consumergate.Key{ConsumerGroup: "test-group"}, true)
+	gate.setErr(fmt.Errorf("gate medium unavailable"))
+
+	handledMsg := ""
+	c, deliveryChan := startGatedConsumer(t, ctrl, gate, func(_ context.Context, delivery consumer.Delivery) error {
+		handledMsg = delivery.Message().ID
+		return nil
+	})
+
+	msg := entityqueue.NewMessage("msg-1", []byte("payload"), "partition1", nil)
+	mockDel, done := gatedDelivery(ctrl, msg)
+
+	deliveryChan <- mockDel
+	<-done
+
+	assert.Equal(t, "msg-1", handledMsg, "a broken gate medium must not stall the pipeline")
+	assert.Empty(t, gate.parked)
+
+	require.NoError(t, c.Stop(30000))
+}
diff --git a/platform/extension/consumergate/BUILD.bazel b/platform/extension/consumergate/BUILD.bazel
new file mode 100644
index 00000000..aa818783
--- /dev/null
+++ b/platform/extension/consumergate/BUILD.bazel
@@ -0,0 +1,8 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["consumergate.go"],
+    importpath = "github.com/uber/submitqueue/platform/extension/consumergate",
+    visibility = ["//visibility:public"],
+)
diff --git a/platform/extension/consumergate/README.md b/platform/extension/consumergate/README.md
new file mode 100644
index 00000000..02ee0951
--- /dev/null
+++ b/platform/extension/consumergate/README.md
@@ -0,0 +1,23 @@
+# Consumer Gate Extension
+
+Runtime stop/start of individual queue controllers — for deterministic e2e scenario control and for operational pause of a consuming stage — without stopping the service that hosts them. Design: [doc/rfc/consumer-gate.md](../../../doc/rfc/consumer-gate.md).
+
+## Contract
+
+A gate is identified by a consumer group (every controller subscribes with a unique one, so it is the controller's stable runtime name), optionally narrowed to a single partition. The gate owns both the admission mechanism and the parked-delivery observation records: `Enter` checks a delivery's gate key synchronously, and a blocked `Entry`'s `Wait` records the parked delivery (stamping the entered identity and `ParkedAtMs`) before blocking until the gate opens (stamping `ReleasedAtMs`). Stopping is a barrier, not preemption — a delivery already past its gate is not recalled.
+
+The package defines three interfaces plus the `Config`:
+
+- `Gate` exposes `Enter`, a synchronous check keyed on consumer group and partition that returns an `Entry` — a future the caller inspects with `Blocked` and, only when blocked, waits on with `Wait`, supplying the parked-delivery content at that point. Polling implementations (see `file/`) re-check gate state on a timer; notification-capable implementations can release the instant the gate opens. Callers that never need gating wire the `noop/` implementation.
+- `Admin` is the write surface tests and tooling use: close a gate, open it, list what a stopped controller is holding.
+
+Parked records are the "observe" half of stop/observe/start: awaiting one is the only way to *know* a stop caught a specific message (as opposed to the message not having arrived yet), and the `ReleasedAtMs` stamp proves the delivery later proceeded.
+
+## Failure posture
+
+An `Enter` or `Wait` that cannot read or record gate state surfaces the error to its caller without further interpretation. What to do with a failed check — for example, letting the delivery through — is the caller's policy, not the gate's.
+
+## Implementations
+
+- [file/](file/) — gate state as plain files in a shared directory. Pausing a controller is writing a small file, resuming is `rm`, inspecting a paused stage is `ls` and `cat`. In the e2e stack the directory is bind-mounted into every service container, so the test process manipulates gates and reads parked records as local files. Enter consults a per-partition cache that avoids a filesystem stat per delivery; a blocked entry polls on a configurable interval.
+- [noop/](noop/) — a no-op gate whose Enter always returns an unblocked Entry, for callers that do not need runtime gating.
diff --git a/platform/extension/consumergate/consumergate.go b/platform/extension/consumergate/consumergate.go
new file mode 100644
index 00000000..a0a68374
--- /dev/null
+++ b/platform/extension/consumergate/consumergate.go
@@ -0,0 +1,163 @@
+// 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 consumergate defines the consumer-gate extension: runtime stop/start
+// of individual queue controllers without stopping the service that hosts them.
+//
+// A gate is keyed by consumer group (the controller's stable runtime name),
+// optionally narrowed to a single partition. Gate.Enter checks a delivery's
+// gate key synchronously and returns an Entry: an open gate admits the
+// delivery immediately, while a closed gate blocks it inside Entry.Wait until
+// the gate opens or the caller's context is cancelled. The gate owns the wait
+// mechanism and the parked-delivery observation records: Wait records the
+// parked delivery before blocking (stamping ParkedAtMs) and stamps
+// ReleasedAtMs when the delivery proceeds.
+//
+// The package holds the contract only: Gate and Entry (the admission
+// interfaces), Admin (the write surface used by tests and tooling), Config,
+// and the Factory interface. Implementations live in subdirectories (see
+// file/, noop/). See doc/rfc/consumer-gate.md for the design.
+package consumergate
+
+//go:generate mockgen -source=consumergate.go -destination=mock/consumergate_mock.go -package=mock
+
+import "context"
+
+// Key identifies a gate: a consumer group, optionally narrowed to one partition.
+type Key struct {
+	// ConsumerGroup is the gated controller's consumer group — its stable runtime name.
+	ConsumerGroup string
+
+	// PartitionKey optionally narrows the gate to a single partition.
+	// Empty gates every partition of the consumer group.
+	PartitionKey string
+}
+
+// Metadata records why a gate was closed, for the operator who finds it later.
+type Metadata struct {
+	// Reason is a human-readable explanation for the closure.
+	Reason string
+
+	// CreatedBy identifies who or what closed the gate.
+	CreatedBy string
+
+	// CreatedAtMs is when the gate was closed (Unix milliseconds).
+	CreatedAtMs int64
+}
+
+// Parked is the record of one gate-blocked delivery. Callers pass the
+// delivery's content fields (Topic, MessageID, Payload, Attempt) to
+// Entry.Wait; the implementation stamps the identity the entry was entered
+// with (ConsumerGroup, PartitionKey) and the timestamps.
+type Parked struct {
+	// ConsumerGroup is the consumer group whose gate is consulted.
+	ConsumerGroup string
+
+	// Topic is the topic key (the stable logical name) the delivery was
+	// consumed from.
+	Topic string
+
+	// MessageID is the queue message ID of the delivery.
+	MessageID string
+
+	// PartitionKey is the partition the delivery belongs to.
+	PartitionKey string
+
+	// Payload is the message payload, recorded so an observer can assert on it.
+	Payload []byte
+
+	// Attempt is the delivery attempt the message is on.
+	Attempt int
+
+	// ParkedAtMs is when the delivery was parked (Unix milliseconds). Stamped
+	// by the gate implementation when it actually blocks; callers leave it zero.
+	ParkedAtMs int64
+
+	// ReleasedAtMs is when the parked delivery proceeded into the controller
+	// (Unix milliseconds). Stamped by the gate implementation when the gate
+	// opens; zero while still parked. Callers leave it zero.
+	ReleasedAtMs int64
+}
+
+// Gate admits deliveries past their gates. Implementations must be safe for
+// concurrent use.
+type Gate interface {
+	// Enter checks the gate identified by key — the delivery's consumer group
+	// and partition — and returns synchronously. When the gate is open, the
+	// returned Entry is unblocked and the delivery may proceed at once; no
+	// other input is needed on that path. When the gate is closed, the
+	// returned Entry is blocked and holds the delivery until its Wait
+	// returns.
+	//
+	// An error reports that gate state could not be read, without further
+	// interpretation — what to do with a failed check is the caller's policy.
+	Enter(ctx context.Context, key Key) (Entry, error)
+}
+
+// Entry is the future returned by Gate.Enter for one delivery.
+type Entry interface {
+	// Blocked reports whether the gate was closed when the delivery entered.
+	// Wait on an unblocked Entry returns nil immediately.
+	Blocked() bool
+
+	// Wait records the parked delivery (stamping its identity and ParkedAtMs)
+	// and then blocks until the gate opens; the implementation owns the wait
+	// mechanism (a polling implementation re-reads gate state on a timer; a
+	// notification-capable implementation can release the instant the gate
+	// opens) and stamps the record's release when it returns.
+	//
+	// Returns ctx.Err() when ctx is cancelled while blocked. Any other error
+	// reports that gate state could not be read or the record could not be
+	// written, without further interpretation — what to do with a failed
+	// wait is the caller's policy.
+	Wait(ctx context.Context, parked Parked) error
+}
+
+// Admin is the write surface used by tests and tooling to operate gates and
+// inspect what a stopped controller is holding.
+type Admin interface {
+	// Close closes the gate for the key. Closing an already-closed gate
+	// overwrites its metadata.
+	Close(ctx context.Context, key Key, meta Metadata) error
+
+	// Open opens the gate for the key. Opening an already-open gate is a no-op.
+	Open(ctx context.Context, key Key) error
+
+	// ListParked returns every parked-delivery record for the consumer group,
+	// including records already stamped released. Callers filter by topic,
+	// message ID, or release state.
+	ListParked(ctx context.Context, consumerGroup string) ([]Parked, error)
+}
+
+// Config holds the knobs for polling-based gate implementations.
+type Config struct {
+	// PollIntervalMs is the cadence at which polling implementations re-read
+	// gate state (milliseconds). Notification-capable implementations may
+	// ignore it.
+	PollIntervalMs int64
+}
+
+// DefaultConfig returns the default gate configuration: 1s poll interval.
+func DefaultConfig() Config {
+	return Config{
+		PollIntervalMs: 1000,
+	}
+}
+
+// Factory creates Gate instances for dependency injection. Factory
+// implementations live in the wiring layer, not in this package.
+type Factory interface {
+	// For returns a Gate for the given configuration.
+	For(cfg Config) (Gate, error)
+}
diff --git a/platform/extension/consumergate/file/BUILD.bazel b/platform/extension/consumergate/file/BUILD.bazel
new file mode 100644
index 00000000..a37f3011
--- /dev/null
+++ b/platform/extension/consumergate/file/BUILD.bazel
@@ -0,0 +1,20 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["store.go"],
+    importpath = "github.com/uber/submitqueue/platform/extension/consumergate/file",
+    visibility = ["//visibility:public"],
+    deps = ["//platform/extension/consumergate:go_default_library"],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["store_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/extension/consumergate:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+    ],
+)
diff --git a/platform/extension/consumergate/file/README.md b/platform/extension/consumergate/file/README.md
new file mode 100644
index 00000000..31626803
--- /dev/null
+++ b/platform/extension/consumergate/file/README.md
@@ -0,0 +1,21 @@
+# File-Backed Consumer Gate
+
+Stores gate state as plain files under a configured root directory. Presence of a gate file means the gate is closed; deleting it opens the gate. See the [extension README](../README.md) and [doc/rfc/consumer-gate.md](../../../../doc/rfc/consumer-gate.md) for the contract and design rationale.
+
+## Layout
+
+```
+{dir}/gates/{consumer_group}/all                         # gates every partition of the controller
+{dir}/gates/{consumer_group}/p-{urlenc(partition)}       # gates one partition
+{dir}/parked/{consumer_group}/{topic}/{urlenc(id)}.json  # one parked delivery record
+```
+
+Partition keys and message IDs may contain `/` (request IDs like `queue/1`), so they are URL-encoded in file names. Gate files carry human-readable JSON metadata (`reason`, `created_by`, `created_at_ms`); 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.
+
+## Operating it by hand
+
+Pause a controller: write any JSON to `{dir}/gates/{group}/all`. Resume: `rm` the file. Inspect what a paused stage is holding: `ls`/`cat` under `{dir}/parked/{group}/`.
+
+## Reach and limits
+
+The directory is trivially shareable out of process — the e2e stack bind-mounts a host directory into every service container, and the test manipulates gates and reads parked records as local files. A file gates only the replicas that see the directory; fleet-wide pause needs the deployment platform to distribute the file, or a store-backed implementation of the same contract.
diff --git a/platform/extension/consumergate/file/store.go b/platform/extension/consumergate/file/store.go
new file mode 100644
index 00000000..04ddb73c
--- /dev/null
+++ b/platform/extension/consumergate/file/store.go
@@ -0,0 +1,417 @@
+// 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 file implements the consumergate contract with plain files in a
+// shared directory. Presence of a gate file means the gate is closed; deleting
+// the file opens it. Layout under the configured root:
+//
+//	gates/{consumer_group}/all                        gates every partition
+//	gates/{consumer_group}/p-{urlenc(partition)}      gates one partition
+//	parked/{consumer_group}/{topic}/{urlenc(id)}.json one parked delivery record
+//
+// Consumer groups and topics are 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 hold human-readable JSON
+// metadata so an operator finding a paused controller can tell why. All writes
+// go through temp-file-plus-rename so readers never see partial JSON.
+//
+// Enter consults a per-(consumerGroup, partitionKey) cached verdict with a TTL
+// equal to the poll interval, so an open gate costs at most one directory stat
+// per partition per interval, not per delivery. A blocked Entry's Wait writes
+// the parked record before blocking, then polls on a ticker at the poll
+// interval, re-reading gate state directly and updating the cache on each
+// tick. When the gate opens, Wait stamps ReleasedAtMs on the record and
+// returns nil.
+//
+// The medium is deliberately the simplest that satisfies the contract: pausing
+// a controller is writing a small file, resuming is rm, inspecting a paused
+// stage is ls and cat, and a bind mount makes all of it reachable from outside
+// the service process. A file gates only the replicas that see the directory —
+// fleet-wide coordination belongs to a future store-backed implementation of
+// the same contract.
+package file
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/uber/submitqueue/platform/extension/consumergate"
+)
+
+// Store implements consumergate.Gate and consumergate.Admin over a directory.
+type Store struct {
+	dir          string
+	pollInterval time.Duration
+
+	// cache holds the last gate verdict per (consumer group, partition),
+	// refreshed at most once per pollInterval. Wait therefore does not hit
+	// the filesystem per message when the gate is open.
+	mu    sync.Mutex
+	cache map[cacheKey]cacheEntry
+}
+
+// cacheKey identifies one cached gate verdict.
+type cacheKey struct {
+	consumerGroup string
+	partitionKey  string
+}
+
+// cacheEntry is a cached gate verdict and when it was read.
+type cacheEntry struct {
+	gated bool
+	at    time.Time
+}
+
+// Verify interface compliance at compile time.
+var (
+	_ consumergate.Gate  = (*Store)(nil)
+	_ consumergate.Admin = (*Store)(nil)
+)
+
+// New returns a file-backed consumergate store rooted at dir. The directory
+// does not need to exist yet — reads treat a missing tree as "no gates, no
+// parked records", and writes create what they need. cfg.PollIntervalMs
+// controls how often a blocked delivery re-checks gate state; values <= 0
+// fall back to the default (1s).
+func New(dir string, cfg consumergate.Config) *Store {
+	if cfg.PollIntervalMs <= 0 {
+		cfg.PollIntervalMs = consumergate.DefaultConfig().PollIntervalMs
+	}
+	return &Store{
+		dir:          dir,
+		pollInterval: time.Duration(cfg.PollIntervalMs) * time.Millisecond,
+		cache:        make(map[cacheKey]cacheEntry),
+	}
+}
+
+// gatePath returns the gate file path for a key: the "all" marker when the key
+// has no partition, or the partition-scoped "p-..." marker otherwise.
+func (s *Store) gatePath(key consumergate.Key) string {
+	name := "all"
+	if key.PartitionKey != "" {
+		name = "p-" + url.QueryEscape(key.PartitionKey)
+	}
+	return filepath.Join(s.dir, "gates", key.ConsumerGroup, name)
+}
+
+// parkedPath returns the parked-record file path for one delivery.
+func (s *Store) parkedPath(consumerGroup, topic, messageID string) string {
+	return filepath.Join(s.dir, "parked", consumerGroup, topic, url.QueryEscape(messageID)+".json")
+}
+
+// isGated reports whether deliveries for the consumer group and partition are
+// currently gated, either by an all-partitions gate or by a gate scoped to
+// exactly this partition.
+func (s *Store) isGated(consumerGroup, partitionKey string) (bool, error) {
+	paths := []string{s.gatePath(consumergate.Key{ConsumerGroup: consumerGroup})}
+	if partitionKey != "" {
+		paths = append(paths, s.gatePath(consumergate.Key{ConsumerGroup: consumerGroup, PartitionKey: partitionKey}))
+	}
+	for _, p := range paths {
+		switch _, err := os.Stat(p); {
+		case err == nil:
+			return true, nil
+		case os.IsNotExist(err):
+			// Not gated by this marker; check the next.
+		default:
+			return false, fmt.Errorf("failed to stat gate file %s: %w", p, err)
+		}
+	}
+	return false, nil
+}
+
+// cachedIsGated returns the cached gate verdict for the group/partition,
+// refreshing it from the filesystem when older than the poll interval.
+func (s *Store) cachedIsGated(consumerGroup, partitionKey string) (bool, error) {
+	key := cacheKey{consumerGroup: consumerGroup, partitionKey: partitionKey}
+	now := time.Now()
+
+	s.mu.Lock()
+	entry, ok := s.cache[key]
+	s.mu.Unlock()
+	if ok && now.Sub(entry.at) < s.pollInterval {
+		return entry.gated, nil
+	}
+
+	gated, err := s.isGated(consumerGroup, partitionKey)
+	if err != nil {
+		return false, err
+	}
+
+	s.mu.Lock()
+	s.cache[key] = cacheEntry{gated: gated, at: now}
+	s.mu.Unlock()
+
+	return gated, nil
+}
+
+// updateCache writes a fresh verdict into the cache.
+func (s *Store) updateCache(consumerGroup, partitionKey string, gated bool) {
+	key := cacheKey{consumerGroup: consumerGroup, partitionKey: partitionKey}
+	s.mu.Lock()
+	s.cache[key] = cacheEntry{gated: gated, at: time.Now()}
+	s.mu.Unlock()
+}
+
+// Enter implements consumergate.Gate. It returns an unblocked Entry when the
+// gate identified by key is open, and a blocked Entry — whose Wait records the
+// parked delivery and polls for the gate to open — when it is closed.
+func (s *Store) Enter(_ context.Context, key consumergate.Key) (consumergate.Entry, error) {
+	gated, err := s.cachedIsGated(key.ConsumerGroup, key.PartitionKey)
+	if err != nil {
+		return nil, err
+	}
+	if !gated {
+		return openEntry{}, nil
+	}
+	return &parkedEntry{store: s, key: key}, nil
+}
+
+// openEntry is the Entry for a delivery that cleared an open gate.
+type openEntry struct{}
+
+// Blocked implements consumergate.Entry.
+func (openEntry) Blocked() bool { return false }
+
+// Wait implements consumergate.Entry. An open gate never blocks and records
+// nothing.
+func (openEntry) Wait(context.Context, consumergate.Parked) error { return nil }
+
+// parkedEntry is the Entry for a delivery held by a closed gate.
+type parkedEntry struct {
+	// store is the file store that gated the delivery.
+	store *Store
+	// key is the gate identity the delivery entered with.
+	key consumergate.Key
+}
+
+// Blocked implements consumergate.Entry.
+func (*parkedEntry) Blocked() bool { return true }
+
+// Wait implements consumergate.Entry. It records the parked delivery (stamping
+// the entry's identity and ParkedAtMs) before blocking, then polls on a ticker
+// at the store's poll interval until the gate opens (stamping ReleasedAtMs) or
+// ctx is cancelled (returning ctx.Err()).
+func (e *parkedEntry) Wait(ctx context.Context, parked consumergate.Parked) error {
+	s := e.store
+
+	// The entered key is the identity source of truth for the record; the
+	// caller supplies only the delivery content.
+	parked.ConsumerGroup = e.key.ConsumerGroup
+	parked.PartitionKey = e.key.PartitionKey
+	parked.ParkedAtMs = time.Now().UnixMilli()
+	if err := s.recordParked(parked); err != nil {
+		return err
+	}
+
+	ticker := time.NewTicker(s.pollInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+
+		gated, err := s.isGated(e.key.ConsumerGroup, e.key.PartitionKey)
+		if err != nil {
+			return err
+		}
+		s.updateCache(e.key.ConsumerGroup, e.key.PartitionKey, gated)
+		if !gated {
+			if err := s.recordReleased(e.key.ConsumerGroup, parked.Topic, parked.MessageID, time.Now().UnixMilli()); err != nil {
+				return err
+			}
+			return nil
+		}
+	}
+}
+
+// recordParked writes a parked-delivery record. Re-recording the same delivery
+// (e.g. after a redelivery) overwrites the previous record.
+func (s *Store) recordParked(parked consumergate.Parked) error {
+	path := s.parkedPath(parked.ConsumerGroup, parked.Topic, parked.MessageID)
+	if err := writeJSON(path, parkedRecord(parked)); err != nil {
+		return fmt.Errorf("failed to write parked record %s: %w", path, err)
+	}
+	return nil
+}
+
+// recordReleased stamps ReleasedAtMs on the existing parked record.
+func (s *Store) recordReleased(consumerGroup, topic, messageID string, releasedAtMs int64) error {
+	path := s.parkedPath(consumerGroup, topic, messageID)
+	rec, err := readParked(path)
+	if err != nil {
+		return fmt.Errorf("failed to read parked record %s: %w", path, err)
+	}
+	rec.ReleasedAtMs = releasedAtMs
+	if err := writeJSON(path, rec); err != nil {
+		return fmt.Errorf("failed to write parked record %s: %w", path, err)
+	}
+	return nil
+}
+
+// Close implements consumergate.Admin by writing the gate file for the key.
+func (s *Store) Close(_ context.Context, key consumergate.Key, meta consumergate.Metadata) error {
+	if key.ConsumerGroup == "" {
+		return fmt.Errorf("gate key requires a consumer group")
+	}
+	path := s.gatePath(key)
+	if err := writeJSON(path, gateRecord{
+		Reason:      meta.Reason,
+		CreatedBy:   meta.CreatedBy,
+		CreatedAtMs: meta.CreatedAtMs,
+	}); err != nil {
+		return fmt.Errorf("failed to write gate file %s: %w", path, err)
+	}
+	return nil
+}
+
+// Open implements consumergate.Admin by removing the gate file for the key.
+// Opening an already-open gate is a no-op.
+func (s *Store) Open(_ context.Context, key consumergate.Key) error {
+	path := s.gatePath(key)
+	if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("failed to remove gate file %s: %w", path, err)
+	}
+	return nil
+}
+
+// ListParked implements consumergate.Admin. It returns every parked record for
+// the consumer group across all topics; a missing tree yields an empty list.
+func (s *Store) ListParked(_ context.Context, consumerGroup string) ([]consumergate.Parked, error) {
+	groupDir := filepath.Join(s.dir, "parked", consumerGroup)
+	topics, err := os.ReadDir(groupDir)
+	if os.IsNotExist(err) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("failed to read parked dir %s: %w", groupDir, err)
+	}
+
+	var out []consumergate.Parked
+	for _, topic := range topics {
+		if !topic.IsDir() {
+			continue
+		}
+		topicDir := filepath.Join(groupDir, topic.Name())
+		entries, err := os.ReadDir(topicDir)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read parked dir %s: %w", topicDir, err)
+		}
+		for _, entry := range entries {
+			// Skip anything that is not a finished record (e.g. temp files
+			// awaiting rename).
+			if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".json") {
+				continue
+			}
+			rec, err := readParked(filepath.Join(topicDir, entry.Name()))
+			if err != nil {
+				return nil, err
+			}
+			out = append(out, consumergate.Parked(rec))
+		}
+	}
+	return out, nil
+}
+
+// gateRecord is the JSON layout of a gate file.
+type gateRecord struct {
+	// Reason is a human-readable explanation for the closure.
+	Reason string `json:"reason"`
+	// CreatedBy identifies who or what closed the gate.
+	CreatedBy string `json:"created_by"`
+	// CreatedAtMs is when the gate was closed (Unix milliseconds).
+	CreatedAtMs int64 `json:"created_at_ms"`
+}
+
+// parkedRecord is the JSON layout of a parked-delivery record. It mirrors
+// consumergate.Parked field-for-field; the named type only pins the wire tags.
+type parkedRecord struct {
+	// ConsumerGroup is the consumer group whose gate parked the delivery.
+	ConsumerGroup string `json:"consumer_group"`
+	// Topic is the topic name the delivery was consumed from.
+	Topic string `json:"topic"`
+	// MessageID is the queue message ID of the parked delivery.
+	MessageID string `json:"message_id"`
+	// PartitionKey is the partition the delivery belongs to.
+	PartitionKey string `json:"partition_key"`
+	// Payload is the message payload (base64 in the JSON encoding).
+	Payload []byte `json:"payload"`
+	// Attempt is the delivery attempt the parked message is on.
+	Attempt int `json:"attempt"`
+	// ParkedAtMs is when the delivery was parked (Unix milliseconds).
+	ParkedAtMs int64 `json:"parked_at_ms"`
+	// ReleasedAtMs is when the delivery proceeded; zero while parked.
+	ReleasedAtMs int64 `json:"released_at_ms"`
+}
+
+// readParked loads and decodes one parked-record file.
+func readParked(path string) (parkedRecord, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return parkedRecord{}, fmt.Errorf("failed to read parked record %s: %w", path, err)
+	}
+	var rec parkedRecord
+	if err := json.Unmarshal(data, &rec); err != nil {
+		return parkedRecord{}, fmt.Errorf("failed to decode parked record %s: %w", path, err)
+	}
+	return rec, nil
+}
+
+// writeJSON writes v as indented JSON via temp-file-plus-rename in the target
+// directory, so concurrent readers never observe partial content. On any
+// failure after the temp file is created, the temp file is removed in a single
+// deferred cleanup; removal errors are joined with the causal error.
+func writeJSON(path string, v any) (retErr error) {
+	data, err := json.MarshalIndent(v, "", "  ")
+	if err != nil {
+		return fmt.Errorf("failed to encode %s: %w", path, err)
+	}
+	dir := filepath.Dir(path)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return fmt.Errorf("failed to create dir %s: %w", dir, err)
+	}
+	tmp, err := os.CreateTemp(dir, filepath.Base(path)+".tmp")
+	if err != nil {
+		return fmt.Errorf("failed to create temp file in %s: %w", dir, err)
+	}
+	tmpName := tmp.Name()
+	defer func() {
+		if retErr != nil {
+			if rmErr := os.Remove(tmpName); rmErr != nil && !os.IsNotExist(rmErr) {
+				retErr = errors.Join(retErr, rmErr)
+			}
+		}
+	}()
+	if _, err := tmp.Write(data); err != nil {
+		closeErr := tmp.Close()
+		return errors.Join(fmt.Errorf("failed to write temp file %s: %w", tmpName, err), closeErr)
+	}
+	if err := tmp.Close(); err != nil {
+		return fmt.Errorf("failed to close temp file %s: %w", tmpName, err)
+	}
+	if err := os.Rename(tmpName, path); err != nil {
+		return fmt.Errorf("failed to rename %s to %s: %w", tmpName, path, err)
+	}
+	return nil
+}
diff --git a/platform/extension/consumergate/file/store_test.go b/platform/extension/consumergate/file/store_test.go
new file mode 100644
index 00000000..07e02da4
--- /dev/null
+++ b/platform/extension/consumergate/file/store_test.go
@@ -0,0 +1,326 @@
+// 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 file
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/uber/submitqueue/platform/extension/consumergate"
+)
+
+// testCfg keeps Wait tests fast: 5ms poll interval.
+var testCfg = consumergate.Config{PollIntervalMs: 5}
+
+func TestIsGated(t *testing.T) {
+	ctx := context.Background()
+
+	tests := []struct {
+		name      string
+		close     []consumergate.Key
+		group     string
+		partition string
+		want      bool
+	}{
+		{
+			name:      "no gates",
+			group:     "orchestrator-batch",
+			partition: "queue-a",
+			want:      false,
+		},
+		{
+			name:      "all-partitions gate matches any partition",
+			close:     []consumergate.Key{{ConsumerGroup: "orchestrator-batch"}},
+			group:     "orchestrator-batch",
+			partition: "queue-a",
+			want:      true,
+		},
+		{
+			name:      "all-partitions gate matches empty partition",
+			close:     []consumergate.Key{{ConsumerGroup: "orchestrator-batch"}},
+			group:     "orchestrator-batch",
+			partition: "",
+			want:      true,
+		},
+		{
+			name:      "partition gate matches its partition",
+			close:     []consumergate.Key{{ConsumerGroup: "orchestrator-batch", PartitionKey: "queue-a"}},
+			group:     "orchestrator-batch",
+			partition: "queue-a",
+			want:      true,
+		},
+		{
+			name:      "partition gate leaves other partitions open",
+			close:     []consumergate.Key{{ConsumerGroup: "orchestrator-batch", PartitionKey: "queue-a"}},
+			group:     "orchestrator-batch",
+			partition: "queue-b",
+			want:      false,
+		},
+		{
+			name:      "gate on one group leaves other groups open",
+			close:     []consumergate.Key{{ConsumerGroup: "orchestrator-batch"}},
+			group:     "runway-merge",
+			partition: "queue-a",
+			want:      false,
+		},
+		{
+			name:      "partition key with slash is encoded and matched",
+			close:     []consumergate.Key{{ConsumerGroup: "orchestrator-batch", PartitionKey: "queue/1"}},
+			group:     "orchestrator-batch",
+			partition: "queue/1",
+			want:      true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			store := New(t.TempDir(), testCfg)
+			for _, key := range tt.close {
+				require.NoError(t, store.Close(ctx, key, consumergate.Metadata{Reason: "test", CreatedBy: "unit", CreatedAtMs: 1}))
+			}
+			got, err := store.isGated(tt.group, tt.partition)
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
+
+func TestOpenClosesGate(t *testing.T) {
+	ctx := context.Background()
+	store := New(t.TempDir(), testCfg)
+	key := consumergate.Key{ConsumerGroup: "orchestrator-batch", PartitionKey: "queue-a"}
+
+	require.NoError(t, store.Close(ctx, key, consumergate.Metadata{Reason: "pause", CreatedBy: "unit", CreatedAtMs: 1}))
+	gated, err := store.isGated(key.ConsumerGroup, key.PartitionKey)
+	require.NoError(t, err)
+	require.True(t, gated)
+
+	require.NoError(t, store.Open(ctx, key))
+	gated, err = store.isGated(key.ConsumerGroup, key.PartitionKey)
+	require.NoError(t, err)
+	assert.False(t, gated)
+
+	// Opening an already-open gate is a no-op.
+	require.NoError(t, store.Open(ctx, key))
+}
+
+func TestCloseRequiresConsumerGroup(t *testing.T) {
+	store := New(t.TempDir(), testCfg)
+	err := store.Close(context.Background(), consumergate.Key{}, consumergate.Metadata{})
+	require.Error(t, err)
+}
+
+func TestParkedRecordLifecycle(t *testing.T) {
+	ctx := context.Background()
+	store := New(t.TempDir(), testCfg)
+
+	parked := consumergate.Parked{
+		ConsumerGroup: "runway-mergeconflictcheck",
+		Topic:         "merge-conflict-check",
+		MessageID:     "e2e-queue/42",
+		PartitionKey:  "e2e-queue",
+		Payload:       []byte(`{"id":"e2e-queue/42"}`),
+		Attempt:       1,
+		ParkedAtMs:    1111,
+	}
+	require.NoError(t, store.recordParked(parked))
+
+	records, err := store.ListParked(ctx, parked.ConsumerGroup)
+	require.NoError(t, err)
+	require.Len(t, records, 1)
+	assert.Equal(t, parked, records[0])
+	assert.Zero(t, records[0].ReleasedAtMs)
+
+	// Re-recording the same delivery (redelivery) overwrites, not duplicates.
+	parked.Attempt = 2
+	require.NoError(t, store.recordParked(parked))
+	records, err = store.ListParked(ctx, parked.ConsumerGroup)
+	require.NoError(t, err)
+	require.Len(t, records, 1)
+	assert.Equal(t, 2, records[0].Attempt)
+
+	require.NoError(t, store.recordReleased(parked.ConsumerGroup, parked.Topic, parked.MessageID, 2222))
+	records, err = store.ListParked(ctx, parked.ConsumerGroup)
+	require.NoError(t, err)
+	require.Len(t, records, 1)
+	assert.Equal(t, int64(2222), records[0].ReleasedAtMs)
+	assert.Equal(t, parked.Payload, records[0].Payload, "release stamp must preserve the recorded payload")
+}
+
+func TestListParkedEmpty(t *testing.T) {
+	store := New(t.TempDir(), testCfg)
+	records, err := store.ListParked(context.Background(), "no-such-group")
+	require.NoError(t, err)
+	assert.Empty(t, records)
+}
+
+func TestListParkedSkipsTempFiles(t *testing.T) {
+	ctx := context.Background()
+	dir := t.TempDir()
+	store := New(dir, testCfg)
+
+	parked := consumergate.Parked{
+		ConsumerGroup: "group",
+		Topic:         "topic",
+		MessageID:     "id",
+		PartitionKey:  "part",
+		ParkedAtMs:    1,
+	}
+	require.NoError(t, store.recordParked(parked))
+
+	// Simulate an in-flight temp file awaiting rename alongside the record.
+	tmpPath := filepath.Join(dir, "parked", "group", "topic", "id.json.tmp123")
+	require.NoError(t, os.WriteFile(tmpPath, []byte("partial"), 0o644))
+
+	records, err := store.ListParked(ctx, "group")
+	require.NoError(t, err)
+	assert.Len(t, records, 1)
+}
+
+func TestMissingDirIsNotGated(t *testing.T) {
+	store := New(filepath.Join(t.TempDir(), "does-not-exist"), testCfg)
+	gated, err := store.isGated("group", "part")
+	require.NoError(t, err)
+	assert.False(t, gated)
+}
+
+func TestEnter_OpenGateUnblocked(t *testing.T) {
+	ctx := context.Background()
+	store := New(t.TempDir(), testCfg)
+
+	entry, err := store.Enter(ctx, consumergate.Key{ConsumerGroup: "group", PartitionKey: "part"})
+	require.NoError(t, err)
+	assert.False(t, entry.Blocked())
+	require.NoError(t, entry.Wait(ctx, consumergate.Parked{
+		Topic:     "topic",
+		MessageID: "msg-1",
+		Payload:   []byte("hello"),
+		Attempt:   1,
+	}))
+
+	// No parked record should exist — the gate was open.
+	records, err := store.ListParked(ctx, "group")
+	require.NoError(t, err)
+	assert.Empty(t, records)
+}
+
+func TestEnter_ClosedGateParksThenReleases(t *testing.T) {
+	ctx := context.Background()
+	store := New(t.TempDir(), testCfg)
+	key := consumergate.Key{ConsumerGroup: "group"}
+
+	require.NoError(t, store.Close(ctx, key, consumergate.Metadata{Reason: "test", CreatedBy: "unit", CreatedAtMs: 1}))
+
+	entry, err := store.Enter(ctx, consumergate.Key{ConsumerGroup: "group", PartitionKey: "part"})
+	require.NoError(t, err)
+	require.True(t, entry.Blocked())
+
+	// Wait records the parked delivery before blocking; the caller supplies
+	// only the delivery content, the store stamps the entered identity.
+	waitDone := make(chan error, 1)
+	go func() {
+		waitDone <- entry.Wait(ctx, consumergate.Parked{
+			Topic:     "topic",
+			MessageID: "msg-1",
+			Payload:   []byte("hello"),
+			Attempt:   1,
+		})
+	}()
+
+	require.Eventually(t, func() bool {
+		records, listErr := store.ListParked(ctx, "group")
+		return listErr == nil && len(records) == 1
+	}, 2*store.pollInterval, store.pollInterval/2)
+
+	records, err := store.ListParked(ctx, "group")
+	require.NoError(t, err)
+	require.Len(t, records, 1)
+	assert.Equal(t, "group", records[0].ConsumerGroup)
+	assert.Equal(t, "part", records[0].PartitionKey)
+	assert.Equal(t, "msg-1", records[0].MessageID)
+	assert.Equal(t, "topic", records[0].Topic)
+	assert.Equal(t, []byte("hello"), records[0].Payload)
+	assert.Equal(t, 1, records[0].Attempt)
+	assert.NotZero(t, records[0].ParkedAtMs)
+	assert.Zero(t, records[0].ReleasedAtMs)
+
+	// Assert Wait has not returned yet.
+	select {
+	case <-waitDone:
+		t.Fatal("Wait returned before the gate was opened")
+	default:
+	}
+
+	// Open the gate — Wait should return nil and the record should have
+	// ReleasedAtMs stamped.
+	require.NoError(t, store.Open(ctx, key))
+	require.NoError(t, <-waitDone)
+
+	records, err = store.ListParked(ctx, "group")
+	require.NoError(t, err)
+	require.Len(t, records, 1)
+	assert.NotZero(t, records[0].ReleasedAtMs)
+}
+
+func TestEnter_ClosedGateCtxCancel(t *testing.T) {
+	store := New(t.TempDir(), testCfg)
+	key := consumergate.Key{ConsumerGroup: "group"}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	require.NoError(t, store.Close(ctx, key, consumergate.Metadata{Reason: "test", CreatedBy: "unit", CreatedAtMs: 1}))
+
+	entry, err := store.Enter(ctx, consumergate.Key{ConsumerGroup: "group", PartitionKey: "part"})
+	require.NoError(t, err)
+	require.True(t, entry.Blocked())
+
+	waitDone := make(chan error, 1)
+	go func() {
+		waitDone <- entry.Wait(ctx, consumergate.Parked{
+			Topic:     "topic",
+			MessageID: "msg-1",
+			Payload:   []byte("hello"),
+			Attempt:   1,
+		})
+	}()
+
+	// Wait until the parked record appears, then cancel.
+	require.Eventually(t, func() bool {
+		records, listErr := store.ListParked(ctx, "group")
+		return listErr == nil && len(records) == 1
+	}, 2*store.pollInterval, store.pollInterval/2)
+
+	cancel()
+	require.ErrorIs(t, <-waitDone, context.Canceled)
+
+	// The record should remain unreleased.
+	records, err := store.ListParked(context.Background(), "group")
+	require.NoError(t, err)
+	require.Len(t, records, 1)
+	assert.Zero(t, records[0].ReleasedAtMs)
+}
+
+func TestEnter_MediumError(t *testing.T) {
+	// Make the store root a regular file so stat fails with ENOTDIR.
+	dir := filepath.Join(t.TempDir(), "not-a-dir")
+	require.NoError(t, os.WriteFile(dir, []byte("x"), 0o644))
+
+	store := New(dir, testCfg)
+	_, err := store.Enter(context.Background(), consumergate.Key{ConsumerGroup: "group", PartitionKey: "part"})
+	require.Error(t, err)
+}
diff --git a/platform/extension/consumergate/mock/BUILD.bazel b/platform/extension/consumergate/mock/BUILD.bazel
new file mode 100644
index 00000000..f0b94bf2
--- /dev/null
+++ b/platform/extension/consumergate/mock/BUILD.bazel
@@ -0,0 +1,12 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["consumergate_mock.go"],
+    importpath = "github.com/uber/submitqueue/platform/extension/consumergate/mock",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//platform/extension/consumergate:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+    ],
+)
diff --git a/platform/extension/consumergate/mock/consumergate_mock.go b/platform/extension/consumergate/mock/consumergate_mock.go
new file mode 100644
index 00000000..3da1f77a
--- /dev/null
+++ b/platform/extension/consumergate/mock/consumergate_mock.go
@@ -0,0 +1,215 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: consumergate.go
+//
+// Generated by this command:
+//
+//	mockgen -source=consumergate.go -destination=mock/consumergate_mock.go -package=mock
+//
+
+// Package mock is a generated GoMock package.
+package mock
+
+import (
+	context "context"
+	reflect "reflect"
+
+	consumergate "github.com/uber/submitqueue/platform/extension/consumergate"
+	gomock "go.uber.org/mock/gomock"
+)
+
+// MockGate is a mock of Gate interface.
+type MockGate struct {
+	ctrl     *gomock.Controller
+	recorder *MockGateMockRecorder
+	isgomock struct{}
+}
+
+// MockGateMockRecorder is the mock recorder for MockGate.
+type MockGateMockRecorder struct {
+	mock *MockGate
+}
+
+// NewMockGate creates a new mock instance.
+func NewMockGate(ctrl *gomock.Controller) *MockGate {
+	mock := &MockGate{ctrl: ctrl}
+	mock.recorder = &MockGateMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockGate) EXPECT() *MockGateMockRecorder {
+	return m.recorder
+}
+
+// Enter mocks base method.
+func (m *MockGate) Enter(ctx context.Context, key consumergate.Key) (consumergate.Entry, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Enter", ctx, key)
+	ret0, _ := ret[0].(consumergate.Entry)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Enter indicates an expected call of Enter.
+func (mr *MockGateMockRecorder) Enter(ctx, key any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Enter", reflect.TypeOf((*MockGate)(nil).Enter), ctx, key)
+}
+
+// MockEntry is a mock of Entry interface.
+type MockEntry struct {
+	ctrl     *gomock.Controller
+	recorder *MockEntryMockRecorder
+	isgomock struct{}
+}
+
+// MockEntryMockRecorder is the mock recorder for MockEntry.
+type MockEntryMockRecorder struct {
+	mock *MockEntry
+}
+
+// NewMockEntry creates a new mock instance.
+func NewMockEntry(ctrl *gomock.Controller) *MockEntry {
+	mock := &MockEntry{ctrl: ctrl}
+	mock.recorder = &MockEntryMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockEntry) EXPECT() *MockEntryMockRecorder {
+	return m.recorder
+}
+
+// Blocked mocks base method.
+func (m *MockEntry) Blocked() bool {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Blocked")
+	ret0, _ := ret[0].(bool)
+	return ret0
+}
+
+// Blocked indicates an expected call of Blocked.
+func (mr *MockEntryMockRecorder) Blocked() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Blocked", reflect.TypeOf((*MockEntry)(nil).Blocked))
+}
+
+// Wait mocks base method.
+func (m *MockEntry) Wait(ctx context.Context, parked consumergate.Parked) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Wait", ctx, parked)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Wait indicates an expected call of Wait.
+func (mr *MockEntryMockRecorder) Wait(ctx, parked any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Wait", reflect.TypeOf((*MockEntry)(nil).Wait), ctx, parked)
+}
+
+// MockAdmin is a mock of Admin interface.
+type MockAdmin struct {
+	ctrl     *gomock.Controller
+	recorder *MockAdminMockRecorder
+	isgomock struct{}
+}
+
+// MockAdminMockRecorder is the mock recorder for MockAdmin.
+type MockAdminMockRecorder struct {
+	mock *MockAdmin
+}
+
+// NewMockAdmin creates a new mock instance.
+func NewMockAdmin(ctrl *gomock.Controller) *MockAdmin {
+	mock := &MockAdmin{ctrl: ctrl}
+	mock.recorder = &MockAdminMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockAdmin) EXPECT() *MockAdminMockRecorder {
+	return m.recorder
+}
+
+// Close mocks base method.
+func (m *MockAdmin) Close(ctx context.Context, key consumergate.Key, meta consumergate.Metadata) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Close", ctx, key, meta)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Close indicates an expected call of Close.
+func (mr *MockAdminMockRecorder) Close(ctx, key, meta any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockAdmin)(nil).Close), ctx, key, meta)
+}
+
+// ListParked mocks base method.
+func (m *MockAdmin) ListParked(ctx context.Context, consumerGroup string) ([]consumergate.Parked, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ListParked", ctx, consumerGroup)
+	ret0, _ := ret[0].([]consumergate.Parked)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ListParked indicates an expected call of ListParked.
+func (mr *MockAdminMockRecorder) ListParked(ctx, consumerGroup any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListParked", reflect.TypeOf((*MockAdmin)(nil).ListParked), ctx, consumerGroup)
+}
+
+// Open mocks base method.
+func (m *MockAdmin) Open(ctx context.Context, key consumergate.Key) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Open", ctx, key)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Open indicates an expected call of Open.
+func (mr *MockAdminMockRecorder) Open(ctx, key any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Open", reflect.TypeOf((*MockAdmin)(nil).Open), ctx, key)
+}
+
+// MockFactory is a mock of Factory interface.
+type MockFactory struct {
+	ctrl     *gomock.Controller
+	recorder *MockFactoryMockRecorder
+	isgomock struct{}
+}
+
+// MockFactoryMockRecorder is the mock recorder for MockFactory.
+type MockFactoryMockRecorder struct {
+	mock *MockFactory
+}
+
+// NewMockFactory creates a new mock instance.
+func NewMockFactory(ctrl *gomock.Controller) *MockFactory {
+	mock := &MockFactory{ctrl: ctrl}
+	mock.recorder = &MockFactoryMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockFactory) EXPECT() *MockFactoryMockRecorder {
+	return m.recorder
+}
+
+// For mocks base method.
+func (m *MockFactory) For(cfg consumergate.Config) (consumergate.Gate, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "For", cfg)
+	ret0, _ := ret[0].(consumergate.Gate)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// For indicates an expected call of For.
+func (mr *MockFactoryMockRecorder) For(cfg any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockFactory)(nil).For), cfg)
+}
diff --git a/platform/extension/consumergate/noop/BUILD.bazel b/platform/extension/consumergate/noop/BUILD.bazel
new file mode 100644
index 00000000..d4c9c2a0
--- /dev/null
+++ b/platform/extension/consumergate/noop/BUILD.bazel
@@ -0,0 +1,20 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["gate.go"],
+    importpath = "github.com/uber/submitqueue/platform/extension/consumergate/noop",
+    visibility = ["//visibility:public"],
+    deps = ["//platform/extension/consumergate:go_default_library"],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["gate_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/extension/consumergate:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+    ],
+)
diff --git a/platform/extension/consumergate/noop/README.md b/platform/extension/consumergate/noop/README.md
new file mode 100644
index 00000000..62760bc5
--- /dev/null
+++ b/platform/extension/consumergate/noop/README.md
@@ -0,0 +1,3 @@
+# No-op Consumer Gate
+
+A consumergate.Gate whose Enter always returns an unblocked Entry — every delivery flows straight to its controller. Wire it in services and tests that do not need runtime gating.
diff --git a/platform/extension/consumergate/noop/gate.go b/platform/extension/consumergate/noop/gate.go
new file mode 100644
index 00000000..93aa9f69
--- /dev/null
+++ b/platform/extension/consumergate/noop/gate.go
@@ -0,0 +1,49 @@
+// 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 noop provides a no-op consumergate.Gate that admits every delivery
+// immediately. Wire it in services and tests that do not need runtime gating.
+package noop
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/platform/extension/consumergate"
+)
+
+// Verify interface compliance at compile time.
+var (
+	_ consumergate.Gate  = Gate{}
+	_ consumergate.Entry = Gate{}
+)
+
+// Gate is a no-op consumer gate: Enter always returns an unblocked Entry.
+// The same value serves as its own Entry.
+type Gate struct{}
+
+// New returns a no-op Gate.
+func New() Gate {
+	return Gate{}
+}
+
+// Enter implements consumergate.Gate. The delivery is never gated.
+func (g Gate) Enter(_ context.Context, _ consumergate.Key) (consumergate.Entry, error) {
+	return g, nil
+}
+
+// Blocked implements consumergate.Entry. A no-op gate never blocks.
+func (Gate) Blocked() bool { return false }
+
+// Wait implements consumergate.Entry. It returns nil immediately.
+func (Gate) Wait(context.Context, consumergate.Parked) error { return nil }
diff --git a/platform/extension/consumergate/noop/gate_test.go b/platform/extension/consumergate/noop/gate_test.go
new file mode 100644
index 00000000..2b35d9e6
--- /dev/null
+++ b/platform/extension/consumergate/noop/gate_test.go
@@ -0,0 +1,37 @@
+// 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 noop
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/uber/submitqueue/platform/extension/consumergate"
+)
+
+func TestGate_EnterNeverBlocks(t *testing.T) {
+	g := New()
+	entry, err := g.Enter(context.Background(), consumergate.Key{ConsumerGroup: "group", PartitionKey: "part"})
+	require.NoError(t, err)
+	assert.False(t, entry.Blocked())
+	require.NoError(t, entry.Wait(context.Background(), consumergate.Parked{
+		Topic:     "topic",
+		MessageID: "msg-1",
+		Payload:   []byte("hello"),
+		Attempt:   1,
+	}))
+}
diff --git a/platform/metrics/metrics.go b/platform/metrics/metrics.go
index 473e3e27..20b321bf 100644
--- a/platform/metrics/metrics.go
+++ b/platform/metrics/metrics.go
@@ -131,6 +131,13 @@ func NamedHistogram(scope tally.Scope, name string, histogram string, buckets ta
 	return tagged(scope, tags).SubScope(name).Histogram(histogram, buckets)
 }
 
+// NamedLatencyHistogram records a duration on the {name}.{histogram} histogram
+// bucketed with the platform's default latency buckets (sub-millisecond to
+// multi-hour), for percentile distributions of operation durations.
+func NamedLatencyHistogram(scope tally.Scope, name string, histogram string, d time.Duration, tags ...Tag) {
+	tagged(scope, tags).SubScope(name).Histogram(histogram, defaultLatencyBuckets).RecordDuration(d)
+}
+
 // NamedGauge updates the {name}.{gauge} gauge to value. Gauges represent a
 // current point-in-time measurement that can go up or down, such as queue depth,
 // active connections, or in-flight requests.
diff --git a/service/runway/server/BUILD.bazel b/service/runway/server/BUILD.bazel
index 49067b0f..f4efcf0f 100644
--- a/service/runway/server/BUILD.bazel
+++ b/service/runway/server/BUILD.bazel
@@ -12,6 +12,8 @@ go_library(
         "//platform/errs:go_default_library",
         "//platform/errs/generic:go_default_library",
         "//platform/errs/mysql:go_default_library",
+        "//platform/extension/consumergate:go_default_library",
+        "//platform/extension/consumergate/file:go_default_library",
         "//platform/extension/messagequeue:go_default_library",
         "//platform/extension/messagequeue/mysql:go_default_library",
         "//runway/controller:go_default_library",
diff --git a/service/runway/server/main.go b/service/runway/server/main.go
index ee6a0936..6aa3c036 100644
--- a/service/runway/server/main.go
+++ b/service/runway/server/main.go
@@ -34,6 +34,8 @@ import (
 	"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"
 	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
 	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
 	"github.com/uber/submitqueue/runway/controller"
@@ -152,6 +154,7 @@ func run() error {
 			genericerrs.Classifier,
 			mysqlerrs.Classifier,
 		),
+		newConsumerGate(logger),
 	)
 
 	mergerFactory := newMergerFactory()
@@ -288,3 +291,23 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 		},
 	})
 }
+
+// defaultConsumerGateDir is the path the compose stack bind-mounts for gate
+// state files. A missing directory simply means every gate is open.
+const defaultConsumerGateDir = "/var/submitqueue/consumergate"
+
+// getEnv returns environment variable value or default if not set.
+func getEnv(key, defaultVal string) string {
+	if val := os.Getenv(key); val != "" {
+		return val
+	}
+	return defaultVal
+}
+
+// newConsumerGate returns a file-backed consumer gate rooted at the directory
+// from CONSUMER_GATE_DIR (defaulting to defaultConsumerGateDir).
+func newConsumerGate(logger *zap.Logger) consumergate.Gate {
+	dir := getEnv("CONSUMER_GATE_DIR", defaultConsumerGateDir)
+	logger.Info("consumer gate configured", zap.String("dir", dir))
+	return consumergatefile.New(dir, consumergate.DefaultConfig())
+}
diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel
index fa4612ff..1db717d3 100644
--- a/service/stovepipe/server/BUILD.bazel
+++ b/service/stovepipe/server/BUILD.bazel
@@ -11,6 +11,7 @@ go_library(
         "//platform/errs:go_default_library",
         "//platform/errs/generic:go_default_library",
         "//platform/errs/mysql:go_default_library",
+        "//platform/extension/consumergate/noop:go_default_library",
         "//platform/extension/messagequeue:go_default_library",
         "//platform/extension/messagequeue/mysql:go_default_library",
         "//service/stovepipe/server/mapper:go_default_library",
diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go
index 7b3aae31..9c024dc9 100644
--- a/service/stovepipe/server/main.go
+++ b/service/stovepipe/server/main.go
@@ -33,6 +33,7 @@ import (
 	"github.com/uber/submitqueue/platform/errs"
 	genericerrs "github.com/uber/submitqueue/platform/errs/generic"
 	mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql"
+	consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop"
 	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
 	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
 	"github.com/uber/submitqueue/service/stovepipe/server/mapper"
@@ -214,15 +215,18 @@ func run() error {
 	// uses AlwaysRetryableProcessor so every non-nil error from a DLQ controller is
 	// forced retryable — reconciliation must redeliver on any failure because the DLQ
 	// subscription is a final destination (DLQ.Enabled is false on it, so there is no
-	// further DLQ to fall back on).
+	// further DLQ to fall back on). Stovepipe has no gated deployment yet, so both
+	// consumers use the no-op gate.
 	primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry,
 		errs.NewClassifierProcessor(
 			genericerrs.Classifier,
 			mysqlerrs.Classifier,
 		),
+		consumergatenoop.New(),
 	)
 	dlqConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer-dlq"), registry,
 		errs.AlwaysRetryableProcessor,
+		consumergatenoop.New(),
 	)
 
 	processController := process.NewController(
diff --git a/service/submitqueue/docker-compose.yml b/service/submitqueue/docker-compose.yml
index e8d2c4bd..97a80a31 100644
--- a/service/submitqueue/docker-compose.yml
+++ b/service/submitqueue/docker-compose.yml
@@ -7,6 +7,12 @@
 #
 # Quick start:
 #   make e2e-test
+#
+# Consumer gate: every service shares one host directory (bind-mounted at
+# /var/submitqueue/consumergate) so tests and operators can stop/start
+# individual queue controllers by writing/removing gate files from the host.
+# Override the host side with SQ_CONSUMER_GATE_DIR (the e2e suite points it at
+# a per-run temp dir); it defaults to /tmp/sq-consumergate for local runs.
 
 services:
   # Application Database - Stores business data (requests, counters, etc.)
@@ -57,6 +63,10 @@ services:
       - QUEUE_CONFIG_PATH=/root/queues.yaml
       # Stable subscriber name for the request-log consumer
       - HOSTNAME=gateway-dev
+      # Consumer-gate state shared with the host (see header comment)
+      - CONSUMER_GATE_DIR=/var/submitqueue/consumergate
+    volumes:
+      - ${SQ_CONSUMER_GATE_DIR:-/tmp/sq-consumergate}:/var/submitqueue/consumergate
     depends_on:
       mysql-app:
         condition: service_healthy
@@ -76,6 +86,10 @@ services:
       # Queue infrastructure connection (separate database)
       - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true
       - HOSTNAME=orchestrator-dev
+      # Consumer-gate state shared with the host (see header comment)
+      - CONSUMER_GATE_DIR=/var/submitqueue/consumergate
+    volumes:
+      - ${SQ_CONSUMER_GATE_DIR:-/tmp/sq-consumergate}:/var/submitqueue/consumergate
     depends_on:
       mysql-app:
         condition: service_healthy
@@ -98,6 +112,10 @@ services:
       # Queue infrastructure connection (shared with the orchestrator)
       - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true
       - HOSTNAME=runway-dev
+      # Consumer-gate state shared with the host (see header comment)
+      - CONSUMER_GATE_DIR=/var/submitqueue/consumergate
+    volumes:
+      - ${SQ_CONSUMER_GATE_DIR:-/tmp/sq-consumergate}:/var/submitqueue/consumergate
     depends_on:
       mysql-queue:
         condition: service_healthy
diff --git a/service/submitqueue/gateway/server/BUILD.bazel b/service/submitqueue/gateway/server/BUILD.bazel
index 39241b0a..303dd1e8 100644
--- a/service/submitqueue/gateway/server/BUILD.bazel
+++ b/service/submitqueue/gateway/server/BUILD.bazel
@@ -16,6 +16,8 @@ go_library(
         "//platform/errs:go_default_library",
         "//platform/errs/generic:go_default_library",
         "//platform/errs/mysql:go_default_library",
+        "//platform/extension/consumergate:go_default_library",
+        "//platform/extension/consumergate/file:go_default_library",
         "//platform/extension/counter/mysql:go_default_library",
         "//platform/extension/messagequeue:go_default_library",
         "//platform/extension/messagequeue/mysql:go_default_library",
diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go
index 38f3f9be..011f5083 100644
--- a/service/submitqueue/gateway/server/main.go
+++ b/service/submitqueue/gateway/server/main.go
@@ -33,6 +33,8 @@ import (
 	"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"
 	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"
@@ -361,6 +363,7 @@ func run() error {
 			genericerrs.Classifier,
 			mysqlerrs.Classifier,
 		),
+		newConsumerGate(logger),
 	)
 
 	logController := logctrl.NewController(logger.Sugar(), scope, store, topickey.TopicKeyLog, "gateway-log")
@@ -439,3 +442,23 @@ func run() error {
 
 	return err
 }
+
+// defaultConsumerGateDir is the path the compose stack bind-mounts for gate
+// state files. A missing directory simply means every gate is open.
+const defaultConsumerGateDir = "/var/submitqueue/consumergate"
+
+// getEnv returns environment variable value or default if not set.
+func getEnv(key, defaultVal string) string {
+	if val := os.Getenv(key); val != "" {
+		return val
+	}
+	return defaultVal
+}
+
+// newConsumerGate returns a file-backed consumer gate rooted at the directory
+// from CONSUMER_GATE_DIR (defaulting to defaultConsumerGateDir).
+func newConsumerGate(logger *zap.Logger) consumergate.Gate {
+	dir := getEnv("CONSUMER_GATE_DIR", defaultConsumerGateDir)
+	logger.Info("consumer gate configured", zap.String("dir", dir))
+	return consumergatefile.New(dir, consumergate.DefaultConfig())
+}
diff --git a/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel
index 2ccb3493..8e6921e2 100644
--- a/service/submitqueue/orchestrator/server/BUILD.bazel
+++ b/service/submitqueue/orchestrator/server/BUILD.bazel
@@ -17,6 +17,8 @@ go_library(
         "//platform/errs:go_default_library",
         "//platform/errs/generic:go_default_library",
         "//platform/errs/mysql:go_default_library",
+        "//platform/extension/consumergate:go_default_library",
+        "//platform/extension/consumergate/file:go_default_library",
         "//platform/extension/counter:go_default_library",
         "//platform/extension/counter/mysql:go_default_library",
         "//platform/extension/messagequeue:go_default_library",
diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go
index eb1ce9ab..09601f2e 100644
--- a/service/submitqueue/orchestrator/server/main.go
+++ b/service/submitqueue/orchestrator/server/main.go
@@ -37,6 +37,8 @@ import (
 	"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"
 	"github.com/uber/submitqueue/platform/extension/counter"
 	mysqlcounter "github.com/uber/submitqueue/platform/extension/counter/mysql"
 	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
@@ -220,6 +222,11 @@ func run() error {
 	// so every non-nil error from a DLQ controller is forced retryable —
 	// reconciliation must redeliver on any failure because the DLQ
 	// subscriptions are final destinations (there is no further DLQ).
+	// Consumer gate: both consumers are gated uniformly — the gate keys on
+	// consumer group, so a DLQ stage is paused by its own group name just
+	// like a primary stage.
+	gate := newConsumerGate(logger)
+
 	primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry,
 		errs.NewClassifierProcessor(
 			genericerrs.Classifier,
@@ -228,9 +235,11 @@ func run() error {
 			// errors surfaced from either backend.
 			mysqlerrs.Classifier,
 		),
+		gate,
 	)
 	dlqConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer-dlq"), registry,
 		errs.AlwaysRetryableProcessor,
+		gate,
 	)
 
 	// Build the per-queue extension registry: each queue resolves to its own
@@ -772,6 +781,18 @@ func getEnv(key, defaultVal string) string {
 	return defaultVal
 }
 
+// defaultConsumerGateDir is the path the compose stack bind-mounts for gate
+// state files. A missing directory simply means every gate is open.
+const defaultConsumerGateDir = "/var/submitqueue/consumergate"
+
+// newConsumerGate returns a file-backed consumer gate rooted at the directory
+// from CONSUMER_GATE_DIR (defaulting to defaultConsumerGateDir).
+func newConsumerGate(logger *zap.Logger) consumergate.Gate {
+	dir := getEnv("CONSUMER_GATE_DIR", defaultConsumerGateDir)
+	logger.Info("consumer gate configured", zap.String("dir", dir))
+	return consumergatefile.New(dir, consumergate.DefaultConfig())
+}
+
 // parseTimeout parses a duration from environment variable with fallback to default.
 // Returns defaultVal if envVal is empty or cannot be parsed.
 func parseTimeout(envVal string, defaultVal time.Duration) time.Duration {
diff --git a/submitqueue/extension/storage/mock/storage_mock.go b/submitqueue/extension/storage/mock/storage_mock.go
index 6655716f..8e705c1e 100644
--- a/submitqueue/extension/storage/mock/storage_mock.go
+++ b/submitqueue/extension/storage/mock/storage_mock.go
@@ -152,20 +152,6 @@ func (mr *MockStorageMockRecorder) GetRequestStore() *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestStore", reflect.TypeOf((*MockStorage)(nil).GetRequestStore))
 }
 
-// GetSpeculationPathBuildStore mocks base method.
-func (m *MockStorage) GetSpeculationPathBuildStore() storage.SpeculationPathBuildStore {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetSpeculationPathBuildStore")
-	ret0, _ := ret[0].(storage.SpeculationPathBuildStore)
-	return ret0
-}
-
-// GetSpeculationPathBuildStore indicates an expected call of GetSpeculationPathBuildStore.
-func (mr *MockStorageMockRecorder) GetSpeculationPathBuildStore() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetSpeculationPathBuildStore", reflect.TypeOf((*MockStorage)(nil).GetSpeculationPathBuildStore))
-}
-
 // GetRequestSummaryStore mocks base method.
 func (m *MockStorage) GetRequestSummaryStore() storage.RequestSummaryStore {
 	m.ctrl.T.Helper()
@@ -194,6 +180,20 @@ func (mr *MockStorageMockRecorder) GetRequestURIStore() *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestURIStore", reflect.TypeOf((*MockStorage)(nil).GetRequestURIStore))
 }
 
+// GetSpeculationPathBuildStore mocks base method.
+func (m *MockStorage) GetSpeculationPathBuildStore() storage.SpeculationPathBuildStore {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetSpeculationPathBuildStore")
+	ret0, _ := ret[0].(storage.SpeculationPathBuildStore)
+	return ret0
+}
+
+// GetSpeculationPathBuildStore indicates an expected call of GetSpeculationPathBuildStore.
+func (mr *MockStorageMockRecorder) GetSpeculationPathBuildStore() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetSpeculationPathBuildStore", reflect.TypeOf((*MockStorage)(nil).GetSpeculationPathBuildStore))
+}
+
 // GetSpeculationTreeStore mocks base method.
 func (m *MockStorage) GetSpeculationTreeStore() storage.SpeculationTreeStore {
 	m.ctrl.T.Helper()
diff --git a/test/e2e/submitqueue/BUILD.bazel b/test/e2e/submitqueue/BUILD.bazel
index eda42174..bb4cf79f 100644
--- a/test/e2e/submitqueue/BUILD.bazel
+++ b/test/e2e/submitqueue/BUILD.bazel
@@ -18,12 +18,19 @@ go_test(
         "e2e",
         "external",
         "integration",
+        # The suite bind-mounts a host /tmp directory (consumer-gate state) into
+        # the compose services. Bazel's hermetic sandbox /tmp is invisible to
+        # the Docker daemon, so this test must run unsandboxed.
+        "no-sandbox",
     ],
     deps = [
         "//api/base/change/protopb:go_default_library",
         "//api/base/mergestrategy/protopb:go_default_library",
+        "//api/runway/messagequeue:go_default_library",
         "//api/submitqueue/gateway/protopb:go_default_library",
         "//api/submitqueue/orchestrator/protopb:go_default_library",
+        "//platform/extension/consumergate:go_default_library",
+        "//platform/extension/consumergate/file:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/extension/storage/mysql:go_default_library",
diff --git a/test/e2e/submitqueue/harness_test.go b/test/e2e/submitqueue/harness_test.go
index 8a163e32..dbeb20bc 100644
--- a/test/e2e/submitqueue/harness_test.go
+++ b/test/e2e/submitqueue/harness_test.go
@@ -28,12 +28,14 @@ package e2e_test
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	changepb "github.com/uber/submitqueue/api/base/change/protopb"
 	mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb"
 	gatewaypb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb"
+	"github.com/uber/submitqueue/platform/extension/consumergate"
 	"github.com/uber/submitqueue/submitqueue/entity"
 )
 
@@ -130,6 +132,80 @@ func (s *E2EIntegrationSuite) assertStatusesInOrder(sqid string, want ...entity.
 		sqid, want, got)
 }
 
+// assertStatusesNever asserts that none of the banned statuses ever appeared
+// in the GetRequestHistoryByID status timeline.
+func (s *E2EIntegrationSuite) assertStatusesNever(sqid string, banned ...entity.RequestStatus) {
+	t := s.T()
+	got := s.timeline(sqid)
+	for _, b := range banned {
+		assert.NotContainsf(t, got, b,
+			"GetRequestHistoryByID for %s must never contain %q; got %v", sqid, b, got)
+	}
+}
+
+// closeGate closes the consumer gate for the consumer group, scoped to one
+// partition (the queue name for pipeline topics). The gate must be closed
+// before the message that must be caught is published — that makes the stop
+// exact by construction rather than a timing race.
+func (s *E2EIntegrationSuite) closeGate(consumerGroup, partitionKey, reason string) {
+	t := s.T()
+	key := consumergate.Key{ConsumerGroup: consumerGroup, PartitionKey: partitionKey}
+	require.NoError(t, s.gate.Close(s.ctx, key, consumergate.Metadata{
+		Reason:      reason,
+		CreatedBy:   "e2e-suite",
+		CreatedAtMs: time.Now().UnixMilli(),
+	}), "failed to close gate %+v", key)
+	s.log.Logf("Closed consumer gate %s (partition %q)", consumerGroup, partitionKey)
+}
+
+// openGate opens the consumer gate for the consumer group and partition.
+// Opening an already-open gate is a no-op, so it is safe to call from a defer
+// after an explicit open.
+func (s *E2EIntegrationSuite) openGate(consumerGroup, partitionKey string) {
+	t := s.T()
+	key := consumergate.Key{ConsumerGroup: consumerGroup, PartitionKey: partitionKey}
+	require.NoError(t, s.gate.Open(s.ctx, key), "failed to open gate %+v", key)
+	s.log.Logf("Opened consumer gate %s (partition %q)", consumerGroup, partitionKey)
+}
+
+// awaitParked polls the shared gate directory until the delivery identified by
+// (consumer group, topic key, message ID) has a parked record, and returns it.
+// The record is written by the gated service before it blocks, so observing it
+// proves the stopped controller is holding exactly this message — as opposed
+// to the message simply not having arrived yet.
+func (s *E2EIntegrationSuite) awaitParked(consumerGroup, topic, messageID string) consumergate.Parked {
+	return s.awaitParkedRecord(consumerGroup, topic, messageID, false)
+}
+
+// awaitReleased polls until the parked record for the delivery carries a
+// released stamp — proof the parked delivery proceeded into the controller
+// after the gate opened.
+func (s *E2EIntegrationSuite) awaitReleased(consumerGroup, topic, messageID string) consumergate.Parked {
+	return s.awaitParkedRecord(consumerGroup, topic, messageID, true)
+}
+
+// awaitParkedRecord is the shared poll behind awaitParked/awaitReleased.
+func (s *E2EIntegrationSuite) awaitParkedRecord(consumerGroup, topic, messageID string, released bool) consumergate.Parked {
+	t := s.T()
+	var found consumergate.Parked
+	require.Eventually(t, func() bool {
+		records, err := s.gate.ListParked(s.ctx, consumerGroup)
+		if err != nil {
+			s.log.Logf("ListParked(%s) failed, retrying: %v", consumerGroup, err)
+			return false
+		}
+		for _, r := range records {
+			if r.Topic == topic && r.MessageID == messageID && (!released || r.ReleasedAtMs != 0) {
+				found = r
+				return true
+			}
+		}
+		return false
+	}, persistTimeout, persistPollInterval,
+		"delivery %s on topic %s should be parked (released=%v) by gate %s", messageID, topic, released, consumerGroup)
+	return found
+}
+
 // terminalState reads the request's current internal RequestState from the
 // operating store (mysql-app). Unlike the status timeline, RequestState is
 // point-in-time — the Request entity is updated in place under optimistic
diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go
index f23cb7d0..7e20adcd 100644
--- a/test/e2e/submitqueue/suite_test.go
+++ b/test/e2e/submitqueue/suite_test.go
@@ -25,6 +25,7 @@ package e2e_test
 import (
 	"context"
 	"database/sql"
+	"os"
 	"path/filepath"
 	"testing"
 	"time"
@@ -33,8 +34,11 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	"github.com/uber-go/tally"
+	runwaymq "github.com/uber/submitqueue/api/runway/messagequeue"
 	gatewaypb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb"
 	orchestratorpb "github.com/uber/submitqueue/api/submitqueue/orchestrator/protopb"
+	"github.com/uber/submitqueue/platform/extension/consumergate"
+	consumergatefile "github.com/uber/submitqueue/platform/extension/consumergate/file"
 	"github.com/uber/submitqueue/submitqueue/entity"
 	"github.com/uber/submitqueue/submitqueue/extension/storage"
 	storagemysql "github.com/uber/submitqueue/submitqueue/extension/storage/mysql"
@@ -51,9 +55,10 @@ type E2EIntegrationSuite struct {
 	stack              *testutil.ComposeStack
 	gatewayClient      gatewaypb.SubmitQueueGatewayClient
 	orchestratorClient orchestratorpb.SubmitQueueOrchestratorClient
-	db                 *sql.DB              // App database
-	queueDB            *sql.DB              // Queue database
-	requestStore       storage.RequestStore // White-box view of the internal RequestState (app DB)
+	db                 *sql.DB                 // App database
+	queueDB            *sql.DB                 // Queue database
+	requestStore       storage.RequestStore    // White-box view of the internal RequestState (app DB)
+	gate               *consumergatefile.Store // Consumer-gate control plane (shared dir bind-mounted into services)
 }
 
 func TestE2EIntegration(t *testing.T) {
@@ -80,13 +85,31 @@ func (s *E2EIntegrationSuite) SetupSuite() {
 	repoRoot := testutil.FindRepoRoot(t)
 	t.Setenv("REPO_ROOT", repoRoot)
 
+	// Consumer-gate directory, bind-mounted into every service container by the
+	// compose file (SQ_CONSUMER_GATE_DIR → /var/submitqueue/consumergate). The
+	// suite closes/opens gates and reads parked records through the same file
+	// implementation the services use. Created directly under /tmp — not
+	// t.TempDir() — because it must be a host path the Docker daemon can bind
+	// mount, and because the containers write parked records into it as root,
+	// which would make t.TempDir()'s mandatory cleanup fail. Removal is
+	// therefore best-effort.
+	gateDir, err := os.MkdirTemp("/tmp", "sq-consumergate-")
+	require.NoError(t, err, "failed to create consumer-gate dir")
+	t.Cleanup(func() {
+		if rmErr := os.RemoveAll(gateDir); rmErr != nil {
+			s.log.Logf("best-effort consumer-gate dir cleanup failed (root-owned parked records): %v", rmErr)
+		}
+	})
+	t.Setenv("SQ_CONSUMER_GATE_DIR", gateDir)
+	s.gate = consumergatefile.New(gateDir, consumergate.DefaultConfig())
+
 	// Use docker-compose from service/submitqueue (full stack)
 	// NOTE: Assumes Linux binaries are pre-built via make target
 	composeFile := filepath.Join(repoRoot, "service/submitqueue/docker-compose.yml")
 	s.stack = testutil.NewComposeStack(t, s.log, s.ctx, composeFile, "e2e-submitqueue")
 
 	// Start the compose stack (Gateway + Orchestrator + 2 MySQL DBs)
-	err := s.stack.Up()
+	err = s.stack.Up()
 	require.NoError(t, err, "failed to start compose stack")
 
 	s.log.Logf("Compose stack started successfully")
@@ -300,32 +323,76 @@ func (s *E2EIntegrationSuite) TestCancelRequest_InvalidSqid() {
 		"empty sqid should map to InvalidArgument; got %s", st.Code())
 }
 
-// TestCancel_RecordsIntent verifies the deterministic half of the cancel flow:
-// Cancel returns OK and the gateway synchronously records a "cancelling" intent
-// entry in the request_log (written directly to the app DB before the RPC
-// returns, right after the Land "accepted" entry).
+// TestCancel_CaughtPreBatch_NeverLands drives the deterministic cancel
+// scenario from doc/rfc/consumer-gate.md as stop → observe → start: the
+// consumer gate stops runway's merge-conflict-check controller before the
+// request's check message can be answered, so the request is provably held
+// pre-batch while the cancel lands. The change must never reach the repo.
+//
+//  1. Stop: close the gate for runway-mergeconflictcheck, scoped to this
+//     queue's partition, before landing — exact by construction, no timing.
+//  2. Land: the orchestrator runs the request to the merge-conflict-check
+//     hand-off; runway's subscriber delivers the check and the gate parks it.
+//  3. Observe: awaiting the parked record proves the controller is stopped and
+//     holding exactly this request's check (there is otherwise no signal
+//     distinguishing "gated and parked" from "not arrived yet").
+//  4. Act while stopped: cancel the request. It is pre-batch by construction,
+//     so the cancel controller drives it terminal Cancelled directly.
+//  5. Start: open the gate. The parked check proceeds as the same attempt,
+//     runway answers the now-stale check, and the orchestrator drops the
+//     signal for the halted request.
 //
-// It deliberately does NOT assert the terminal "cancelled" outcome. Cancellation
-// is best-effort and races the pipeline: on the hermetic stack the happy path
-// reaches "landed" in ~2s, and a cancel published before the orchestrator's
-// start controller has created the request is rejected to the DLQ and reconciled
-// to "error". Asserting a terminal "cancelled" deterministically needs a
-// pipeline-pause lever (e.g. a runway "park" marker that withholds the
-// merge-conflict-check signal so the request is caught pre-batch) — that is the
-// next incremental, per-stage addition on top of this harness.
-func (s *E2EIntegrationSuite) TestCancel_RecordsIntent() {
+// The drop in step 5 is asserted without sleeping: a sentinel request landed
+// on the same queue after the gate opens shares the check and signal
+// partitions with the stale message, so the sentinel reaching "landed" proves
+// the stale signal was already consumed — at which point the cancelled
+// request must still be terminal Cancelled, never batched, never landed.
+func (s *E2EIntegrationSuite) TestCancel_CaughtPreBatch_NeverLands() {
 	t := s.T()
 
-	sqid := s.land("e2e-cancel-queue", "github://github.example.com/uber/e2e-cancel/pull/9999/abcdef0123456789abcdef0123456789abcdef01")
-	s.log.Logf("Land (cancel path) succeeded: sqid=%s; cancelling", sqid)
+	const queue = "e2e-cancel-queue"
+	const gateGroup = "runway-mergeconflictcheck"
+	gateTopic := runwaymq.TopicKeyMergeConflictCheck.String()
+
+	s.closeGate(gateGroup, queue, "e2e: hold merge-conflict check to catch cancel pre-batch")
+	// Reopen even if an assertion below fails, so teardown does not stop the
+	// stack with a delivery still parked. Opening twice is a no-op.
+	defer s.openGate(gateGroup, queue)
+
+	sqid := s.land(queue, "github://github.example.com/uber/e2e-cancel/pull/9999/abcdef0123456789abcdef0123456789abcdef01")
+	s.log.Logf("Land (cancel path) succeeded: sqid=%s; awaiting parked check", sqid)
+
+	parked := s.awaitParked(gateGroup, gateTopic, sqid)
+	assert.Equal(t, queue, parked.PartitionKey, "check message should be partitioned by queue")
+	assert.NotEmpty(t, parked.Payload, "parked record should carry the check payload")
 
+	// The controller is provably stopped and holding this request's check;
+	// cancel now. The request cannot be batched until the check is answered,
+	// so the cancel controller takes the not-batched path to terminal
+	// Cancelled.
 	_, err := s.gatewayClient.Cancel(s.ctx, &gatewaypb.CancelRequest{Sqid: sqid, Reason: "e2e cancel test"})
 	require.NoError(t, err, "Cancel failed")
 
-	// The gateway writes "accepted" on Land and "cancelling" on Cancel
-	// synchronously, so GetRequestHistoryByID exposes both when Cancel returns.
+	s.awaitStatus(sqid, entity.RequestStatusCancelled)
 	s.assertStatusesInOrder(sqid,
 		entity.RequestStatusAccepted,
 		entity.RequestStatusCancelling,
+		entity.RequestStatusCancelled,
 	)
+	assert.Equal(t, entity.RequestStateCancelled, s.terminalState(sqid),
+		"operating store should show request %s terminal cancelled while its check is parked", sqid)
+
+	// Start the controller again and prove the parked delivery proceeded.
+	s.openGate(gateGroup, queue)
+	s.awaitReleased(gateGroup, gateTopic, sqid)
+
+	// Sentinel on the same queue: its landing proves the stale signal ahead of
+	// it on the same partitions was consumed.
+	sentinel := s.land(queue, "github://github.example.com/uber/e2e-cancel/pull/10000/1234567890abcdef1234567890abcdef12345678")
+	s.awaitStatus(sentinel, entity.RequestStatusLanded)
+
+	// The stale check answer was dropped: the cancelled request never advanced.
+	assert.Equal(t, entity.RequestStateCancelled, s.terminalState(sqid),
+		"request %s must stay terminal cancelled after its stale check signal is processed", sqid)
+	s.assertStatusesNever(sqid, entity.RequestStatusBatched, entity.RequestStatusLanded)
 }
diff --git a/test/integration/submitqueue/core/consumer/BUILD.bazel b/test/integration/submitqueue/core/consumer/BUILD.bazel
index 3afe7e9e..1924a69d 100644
--- a/test/integration/submitqueue/core/consumer/BUILD.bazel
+++ b/test/integration/submitqueue/core/consumer/BUILD.bazel
@@ -15,6 +15,7 @@ go_test(
         "//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/mysql:go_default_library",
         "//test/testutil:go_default_library",
diff --git a/test/integration/submitqueue/core/consumer/consumer_test.go b/test/integration/submitqueue/core/consumer/consumer_test.go
index 98709155..fc558d9f 100644
--- a/test/integration/submitqueue/core/consumer/consumer_test.go
+++ b/test/integration/submitqueue/core/consumer/consumer_test.go
@@ -17,6 +17,7 @@ import (
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/errs"
+	consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop"
 	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
 	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
 	"github.com/uber/submitqueue/test/testutil"
@@ -147,7 +148,7 @@ func (s *ConsumerIntegrationSuite) newConsumer(t *testing.T, q extqueue.Queue, t
 	})
 	require.NoError(t, err)
 
-	return consumer.New(logger, tally.NoopScope, registry, errs.NewClassifierProcessor())
+	return consumer.New(logger, tally.NoopScope, registry, errs.NewClassifierProcessor(), consumergatenoop.New())
 }
 
 func (s *ConsumerIntegrationSuite) TestConsumerPerPartitionIsolation() {