diff --git a/Makefile b/Makefile index 98be4828..7e681ab4 100644 --- a/Makefile +++ b/Makefile @@ -366,7 +366,7 @@ local-stovepipe-stop: ## Stop the Stovepipe service mocks: ## Generate mock files using mockgen @echo "Generating mocks..." - @$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/pusher/... ./submitqueue/extension/scorer/... ./submitqueue/extension/conflict/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/... + @$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/pusher/... ./submitqueue/extension/scorer/... ./submitqueue/extension/conflict/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/... @echo "Mocks generated successfully!" proto: ## Generate protobuf files from .proto definitions diff --git a/doc/rfc/consumer-gate.md b/doc/rfc/consumer-gate.md index e51f3903..777b0bf5 100644 --- a/doc/rfc/consumer-gate.md +++ b/doc/rfc/consumer-gate.md @@ -1,15 +1,15 @@ # Consumer Gate -Stopping and starting individual queue controllers at runtime — for deterministic e2e scenario control and for operational pause of a consuming stage — without stopping the service that hosts them. +Stopping and starting individual queue controllers at runtime for deterministic e2e scenario control and single-host development, without stopping the service that hosts them. ## Problem -The pipeline is a set of queue controllers spread across three services, and several needs reduce to the same primitive: *stop one controller from taking new messages, observe some condition while it is stopped, then start it again.* +The pipeline is a set of queue controllers spread across three services, and deterministic tests need the same primitive repeatedly: *stop one controller from taking new messages, observe some condition while it is stopped, then start it again.* - **E2e scenario control.** A test that must interleave two in-flight messages ("the batch controller must not consume its message before the cancel controller has finished") needs to halt exactly one controller while its siblings keep running. Stopping the whole service is too coarse: it kills every controller in the process, re-assigns ephemeral ports, and turns a scenario step into a container lifecycle event. -- **Operational pause.** During an incident, the safest first move is often "stop consuming" — park a stage that is processing poison input or hammering a struggling dependency — without redeploying or scaling to zero, and while keeping the rest of the service (RPC surface, other controllers) alive. +- **Single-host development.** A developer reproducing an ordering-sensitive scenario locally needs the same stop/observe/start control without restarting the whole stack. -Nothing in the system expresses this today. The queue can be manipulated from outside (e.g. starving a consumer by occupying its partition leases), but data-plane tricks of that kind are structurally limited: they only work if arranged before the controller ever touches the partition, they couple the caller to one backend's internals, and they are invisible to the service — no first-class semantics, no metrics, nothing an operator can reuse. This RFC makes stop/observe/start a small, coherent, first-class mechanism instead. +Nothing in the system expresses this today. The queue can be manipulated from outside (e.g. starving a consumer by occupying its partition leases), but data-plane tricks of that kind are structurally limited: they only work if arranged before the controller ever touches the partition, they couple the caller to one backend's internals, and they are invisible to the service. This RFC makes stop/observe/start a small, coherent mechanism for tests and local development instead. Fleet-wide production pause is explicitly out of scope for the file implementation. ## Decisions @@ -17,7 +17,7 @@ Nothing in the system expresses this today. The queue can be manipulated from ou The consumer framework already owns the two facts that make an in-process gate clean. Dispatch is **serial per partition** — `consumeLoop` routes each delivery to a per-partition goroutine, and the next delivery of a partition is not started until the current one completes — so holding one delivery blocks exactly that partition and nothing else. And the framework **owns ack/nack** — controllers signal outcome only through `Process`'s return value — so a delivery can be held simply by not yet invoking the controller. -The gate is a decorator installed by the consumer around every registered controller. Before invoking `Process`, it consults gate state for the controller's consumer group. If the gate is closed, the delivery is **parked**: the decorator blocks in place, keeping the delivery in-flight and periodically calling `ExtendVisibilityTimeout` — already part of the `Delivery` contract, and specified to *not* increment the retry count — until the gate opens or the consumer shuts down. When the gate opens, the parked delivery proceeds into the controller as the same attempt, in partition order. Parked messages therefore never burn retry budget, never touch the DLQ, and are never lost: if the process dies while parked, extension stops, visibility lapses, and the queue redelivers normally. +The gate is a decorator installed by the consumer around every registered controller. Before invoking `Process`, it consults gate state for the controller's consumer group. If the gate is closed, the delivery is **parked**: the decorator blocks in place, keeping that delivery in flight and periodically calling `ExtendVisibilityTimeout` — already part of the `Delivery` contract, and specified to *not* increment the retry count — until the gate opens or the consumer shuts down. Gating does not acknowledge, nack, reject, remove, or move the source delivery; it remains owned by the queue for the same consumer group and partition. When the gate opens, the same delivery proceeds into the controller in partition order. If the process dies or shuts down while parked, extension stops, visibility lapses, and the queue makes the delivery eligible for normal redelivery. Stopping is a barrier, not preemption: a message already inside `Process` when the gate closes runs to completion; the gate guarantees no *new* message enters the controller. @@ -29,11 +29,11 @@ Every controller subscribes with a unique consumer group (`orchestrator-batch`, ### Gate state is a separate extension -The consumer gate is a shared extension in its own right, not a feature of any queue backend. The contract lives at `platform/extension/consumergate/`: the behavioral interface the middleware reads (is this group/partition gated? record a parked delivery, record its release), the write surface tests and tooling use (close a gate, open it), and the `Config`. Implementations live in subdirectories, per the standard extension layout. The consumer package takes the read-side interface as a dependency — wiring constructs an implementation and passes it to `consumer.New` via a new option; when no gate is configured, the middleware is absent and the consumer behaves exactly as today. The wiring delta is one option argument at each consumer construction site (gateway, orchestrator primary, orchestrator DLQ, runway); no per-controller wiring, and DLQ consumers are gated uniformly with the rest. +The consumer gate is a shared extension in its own right, not a feature of any queue backend. The contract lives at `platform/extension/consumergate/`: the behavioral interface the consumer reads, the write surface tests and tooling use, and the `Config`. `Watch` accepts a caller-owned `DeliveryDescriptor` containing only message data; the implementation combines it with the gate identity captured by `Enter` and its own timestamp to create the observable `Parked` record, so callers cannot supply or overwrite gate-owned fields. Implementations live in subdirectories, per the standard extension layout. The consumer package takes the read-side interface as a dependency; wiring passes the file implementation only when `CONSUMER_GATE_DIR` is explicitly configured and otherwise passes the no-op implementation. Keeping the contract separate from any backend is what lets the storage medium be chosen per deployment: a filesystem directory first (below), a database- or config-service-backed implementation later if fleet-wide coordination demands it — with the middleware, the wiring shape, and every test written against the contract unchanged. -### First implementation: files in a shared directory +### E2E and local implementation: files in a shared directory The first implementation stores gate state as plain files under a configured directory. Presence of a gate file means the gate is closed; deleting the file opens it. Parked deliveries are recorded as JSON files. The layout: @@ -43,29 +43,31 @@ The first implementation stores gate state as plain files under a configured dir {dir}/parked/{consumer_group}/{topic}/{urlenc(id)}.json # one parked delivery record ``` -Consumer groups and topics are already filesystem-safe by the repo's naming rules; partition keys and message IDs may contain `/` (request IDs like `queue/1`), so they are URL-encoded in file names. Gate files contain human-readable JSON metadata — `reason`, `created_by`, `created_at_ms` — so an operator finding a paused controller can tell why. Parked records carry the payload, attempt, `parked_at_ms`, and a `released_at_ms` stamped when the delivery proceeds; all writes go through temp-file-plus-rename so readers never see partial JSON. +Consumer groups and topics are already filesystem-safe by the repo's naming rules; partition keys and message IDs may contain `/` (request IDs like `queue/1`), so they are URL-encoded in file names. Gate files contain human-readable JSON metadata — `reason`, `created_by`, `created_at_ms` — so an operator finding a paused controller can tell why. Parked records carry the payload, attempt, and `parked_at_ms` while a delivery is blocked; the record is deleted before the wait ends, so payloads are not retained after release, cancellation, or monitoring failure. All writes go through temp-file-plus-rename so readers never see partial JSON. -Files are the simplest medium that satisfies every requirement in this RFC, and simplicity is the point of the first implementation: +Files are the simplest medium for the E2E and single-host scope: -- **Operator interface for free.** Pausing a controller is writing a small file; resuming is `rm`. Inspecting a paused stage is `ls` and `cat`. No client, no schema, no query. +- **Developer interface for free.** Pausing a controller is writing a small file; resuming is `rm`. Inspecting a paused stage is `ls` and `cat`. No client, no schema, no query. - **Trivially reachable out of process.** In the e2e stack, the compose file bind-mounts a host directory into every service container at a fixed path (passed via one environment variable); the test process manipulates gates and reads parked records as local files. In single-host dev the same directory works as-is. -- **Durable and independent.** State survives service restarts — a paused stage stays paused until explicitly opened — and the gate has no dependency on the queue backend or any database being healthy. +- **Independent of the queue database.** Gate state remains available while the configured directory remains available. -The middleware **polls** the directory rather than using filesystem notifications: inotify events do not propagate reliably across bind mounts and overlay filesystems, and the cached-poll posture (below) makes notification latency irrelevant. The known limit of the file medium is multi-replica fleets: a file gates the replicas that see the directory, so a fleet-wide pause needs the deployment platform to distribute the file — or a future store-backed implementation of the same contract. That trade is accepted; the deployments this RFC serves (e2e, single-host dev, per-instance operational pause) are exactly where files excel. +The middleware **polls** the directory rather than using filesystem notifications: inotify is platform-specific, watches can overflow or require re-registration, and event behavior varies across bind mounts, overlay or network filesystems, rootless Docker, and Docker Desktop's host/container filesystem bridge. Polling is the portable convergence mechanism; filesystem events may be added later as an optional wakeup optimization alongside it. -### Read path: cached poll, bounded effect latency +The file implementation gates only processes that see the same directory. Its state survives a process or container restart only when that directory is backed by storage that survives the restart, and it does not survive node replacement unless the storage is shared and persistent. It is not a fleet-wide production control plane. Services therefore enable it only through the explicit `CONSUMER_GATE_DIR` opt-in; otherwise they wire the no-op gate. -The middleware does not check gate state per message. Gate state is cached per controller and refreshed on a short interval (configurable, ~1s), and a parked delivery re-checks on the same tick. The dormant cost of the feature — the common case, forever — is one directory stat per controller per interval. The price is that closing a gate takes effect within one refresh interval plus the in-flight message's completion; opening one takes effect within one interval. +### Read path: direct reads and bounded release latency + +The middleware checks the applicable gate files for every delivery. A parked delivery re-checks them on a short interval (configurable, ~1s). Closing a gate therefore affects the next delivery check without waiting for a cache refresh; opening one releases already parked deliveries within one poll interval. Tests do not depend on that latency. The deterministic patterns are two: **arrange first** (close the gate before publishing the message that must be caught — exact by construction), or **await the observed effect** (the parked record, below) instead of assuming timing. ### Observation: parked deliveries are recorded -Parking writes the parked record before blocking. This record is the "observe" half of stop/observe/start: a test awaits the record to *know* the stop caught its message (there is otherwise no signal distinguishing "gated and parked" from "not arrived yet"), can assert on the recorded payload, and can decide what to do next while the controller is provably stopped. For an operator, the same records answer "what is this paused controller holding?". Records are bounded by parked messages, which are bounded by gate usage; the directory is empty whenever no gate is in use. +Parking writes the parked record before blocking. This record is the "observe" half of stop/observe/start: a test awaits the record to *know* the stop caught its message (there is otherwise no signal distinguishing "gated and parked" from "not arrived yet"), can assert on the recorded payload, and can decide what to do next while the controller is provably stopped. The record is removed before the wait reports release, cancellation, or failure, so records are bounded by currently parked messages and the directory is empty whenever no delivery is held behind a gate. ### Failure posture: fail open -If gate state cannot be read (directory missing, I/O error), the middleware logs, increments an error counter, and lets deliveries through. Gating is auxiliary; a broken gate medium must not become a pipeline stall. The consequence — a closed gate is best-effort under infra failure — is acceptable because tests assert observed effects (parked records, downstream state), not the mechanism, and an operator gets the failure loudly in metrics. +If gate state cannot be read (directory missing, I/O error), the middleware logs, increments an error counter, and lets deliveries through. Gating is auxiliary; a broken gate medium must not become a pipeline stall. The consequence — a closed gate is best-effort under infra failure — is acceptable because tests assert observed effects (parked records, downstream state), not the mechanism. ## Test walk-through @@ -82,8 +84,8 @@ The two-controller interleaving scenario is the same shape: close controller X's ## Rejected - **Starving a controller through the queue's data plane** (e.g. occupying its partition leases from outside the service). Needs zero service changes, which is why it was the harness's first candidate, but it is pre-hold-only (an actively consuming controller cannot be stopped), coupled to one backend's scheduling internals, and invisible to the service — no observation, no metrics, nothing reusable by an operator. Once service modifications are on the table, the middleware dominates it. -- **A database-backed store as the first implementation** (a table in the queue database). Reaches every replica and centralizes fleet-wide state, but costs a schema, a store, and migrations for what a directory of files does in the deployments at hand — and it ties gate availability to a database. It remains the natural second implementation of the same extension contract if fleet-wide coordination is ever needed. -- **Admin RPC on each service for pause/resume.** Reaches the same middleware, but costs a new proto surface, port, and auth story on three services, and state vanishes on restart. A shared medium gives out-of-process reach, durability, and one lever shared by tests and operators. +- **A database-backed store in this E2E-focused change.** A shared store is the appropriate direction if fleet-wide production pause is required, but it needs its own availability, caching, and administration design. It should be added as another `consumergate` implementation rather than coupled to a queue backend. +- **Admin RPC on each service for pause/resume.** Reaches the same middleware, but costs a new proto surface, port, and auth story on three services. The shared test directory already supplies the out-of-process control required here. - **Config/env-driven controller enablement plus restart.** Restart granularity is the whole process — it bounces every sibling controller and disturbs in-flight leases, destroying exactly the "others keep running" property mid-scenario. A static topology tool, not a stop/start lever. - **Fail closed on gate-state errors.** Turns an auxiliary medium's failure into a full consumption stall across every gated consumer. The gate exists to *add* control, not to add a new way for the pipeline to stop on its own. - **Message-level breakpoint rules in v1** (match on message ID or payload, single-step release). The parked records and the gate key structure leave room for this evolution, but stop/observe/start at controller and partition granularity covers every scenario currently in hand; rule matching, rule versioning, and partial release are complexity deferred until a test needs them. diff --git a/platform/consumer/BUILD.bazel b/platform/consumer/BUILD.bazel index 30036c30..9214f471 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", @@ -29,6 +30,8 @@ go_test( deps = [ "//platform/base/messagequeue: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", "@com_github_stretchr_testify//assert:go_default_library", diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index 2f684e73..01fb6a63 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 @@ -87,13 +104,20 @@ type activeSubscription struct { // without introducing duplicate consumer sub-scopes. 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, - registry: registry, - processor: processor, - subscriptions: make(map[TopicKey]*activeSubscription), + logger: logger, + metricsScope: scope, + registry: registry, + processor: processor, + gate: gate, + gateExtendInterval: defaultGateExtendInterval, + subscriptions: make(map[TopicKey]*activeSubscription), } } @@ -343,6 +367,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 + } + msg := delivery.Message() topicKey := controller.TopicKey() @@ -466,6 +498,103 @@ 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 left in-flight without processing or ack/nack. +// +// Gate.Enter checks the gate synchronously; an unblocked entry is the common +// path and costs nothing further. For a blocked entry the gate hands back a +// watch channel (its own monitoring goroutine behind it), and this routine +// multiplexes the watch with visibility extension. The source delivery remains +// owned by the queue throughout the wait: gating never acknowledges, rejects, +// nacks, or moves it. On shutdown, extension stops and normal queue visibility +// semantics make the delivery eligible for redelivery. +// +// Failures fail open: if gate state cannot be read or recorded, or the delivery +// can no longer be held safely because visibility extension failed, processing +// proceeds and the failure is surfaced via logs and metrics. +func (m *consumer) waitGate(ctx context.Context, controller Controller, delivery extqueue.Delivery, scope tally.Scope) bool { + const opName = "gate" + + msg := delivery.Message() + consumerGroup := controller.ConsumerGroup() + topic := controller.TopicKey().String() + + entry, err := m.gate.Enter(ctx, consumergate.Key{ConsumerGroup: consumerGroup, PartitionKey: msg.PartitionKey}) + if err != nil { + if errors.Is(err, context.Canceled) { + return false + } + metrics.NamedCounter(scope, opName, "enter_errors", 1) + m.logger.Errorw("gate check failed, failing open", + "consumer_group", consumerGroup, + "topic", topic, + "message_id", msg.ID, + "error", err, + ) + return true + } + if !entry.Blocked() { + return true + } + + start := time.Now() + defer func() { + metrics.NamedHistogram(scope, opName, "wait_latency", metrics.LongLatencyBuckets).RecordDuration(time.Since(start)) + }() + + descriptor := consumergate.DeliveryDescriptor{ + Topic: topic, + MessageID: msg.ID, + Payload: msg.Payload, + Attempt: delivery.Attempt(), + } + + watchCtx, cancelWatch := context.WithCancel(ctx) + defer cancelWatch() + watchCh := entry.Watch(watchCtx, descriptor) + + ticker := time.NewTicker(m.gateExtendInterval) + defer ticker.Stop() + + for { + select { + case waitErr := <-watchCh: + if waitErr == nil { + return true + } + if errors.Is(waitErr, context.Canceled) { + return false + } + 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", waitErr, + ) + return true + + case <-ticker.C: + if extendErr := delivery.ExtendVisibilityTimeout(ctx, gateExtensionMs); extendErr != nil { + cancelWatch() + <-watchCh + if errors.Is(extendErr, context.Canceled) { + return false + } + metrics.NamedCounter(scope, opName, "wait_errors", 1) + m.logger.Errorw("gate visibility extension failed, failing open", + "consumer_group", consumerGroup, + "topic", topic, + "message_id", msg.ID, + "error", extendErr, + ) + return true + } + } + } +} + func controllerClassificationTags(err error) []metrics.Tag { origin := "infra" if errs.IsRetryable(err) { diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go index e141ce08..3e0331da 100644 --- a/platform/consumer/consumer_test.go +++ b/platform/consumer/consumer_test.go @@ -28,6 +28,8 @@ import ( "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" extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "go.uber.org/mock/gomock" @@ -130,7 +132,7 @@ func TestNew(t *testing.T) { reg, err := NewTopicRegistry(nil) require.NoError(t, err) - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) require.NotNil(t, c) } @@ -138,7 +140,7 @@ func TestConsumer_Register(t *testing.T) { logger := zaptest.NewLogger(t).Sugar() reg, _ := NewTopicRegistry(nil) - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) handler1 := &testController{} setupController(handler1, "handler1", testTopicKeyStart, "group1", nil) @@ -157,7 +159,7 @@ func TestConsumer_Register_DuplicateTopic(t *testing.T) { logger := zaptest.NewLogger(t).Sugar() reg, _ := NewTopicRegistry(nil) - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) handler1 := &testController{} setupController(handler1, "handler1", testTopicKeyStart, "group1", nil) @@ -176,7 +178,7 @@ func TestConsumer_Register_AfterStop(t *testing.T) { logger := zaptest.NewLogger(t).Sugar() reg, _ := NewTopicRegistry(nil) - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) err := c.Stop(1000) require.NoError(t, err) @@ -192,7 +194,7 @@ func TestConsumer_Start_NoHandlers(t *testing.T) { logger := zaptest.NewLogger(t).Sugar() reg, _ := NewTopicRegistry(nil) - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) err := c.Start(context.Background()) assert.Error(t, err) @@ -202,7 +204,7 @@ func TestConsumer_Start_AfterStop(t *testing.T) { logger := zaptest.NewLogger(t).Sugar() reg, _ := NewTopicRegistry(nil) - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) handler := &testController{} setupController(handler, "handler1", testTopicKeyStart, "group1", nil) @@ -228,7 +230,7 @@ func TestConsumer_Start_MissingSubscriptionConfig(t *testing.T) { ) require.NoError(t, err) - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) handler := &testController{} setupController(handler, "handler", testTopicKeyStart, "group", nil) @@ -254,7 +256,7 @@ func TestConsumer_Start_SubscribeFailure(t *testing.T) { reg := newRegistry(t, mockQ, testTopicKeyStart, "group") - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) handler := &testController{} setupController(handler, "handler", testTopicKeyStart, "group", nil) @@ -280,7 +282,7 @@ func TestConsumer_ProcessDelivery_Success(t *testing.T) { reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) handledMsg := "" handler := &testController{} @@ -326,7 +328,7 @@ func TestConsumer_ProcessDelivery_Error(t *testing.T) { reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) handler := &testController{} setupController(handler, "test-handler", testTopicKeyStart, "test-group", @@ -368,7 +370,7 @@ func TestConsumer_ProcessDelivery_NonRetryableError(t *testing.T) { reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) handler := &testController{} setupController(handler, "test-handler", testTopicKeyStart, "test-group", @@ -419,7 +421,7 @@ func TestConsumer_Stop(t *testing.T) { reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) handler := &testController{} setupController(handler, "test-handler", testTopicKeyStart, "test-group", nil) @@ -511,7 +513,7 @@ func TestConsumer_ObservabilityTags(t *testing.T) { reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") - testC := New(logger, testScope, reg, tt.processor) + testC := New(logger, testScope, reg, tt.processor, consumergatenoop.New()) handler := &testController{} setupController(handler, "test-handler", testTopicKeyStart, "test-group", @@ -651,7 +653,7 @@ func TestConsumer_AckLifecycleMetrics(t *testing.T) { reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") - c := New(logger, scope, reg, errs.NewClassifierProcessor()) + c := New(logger, scope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) handler := &testController{} setupController(handler, "test-handler", testTopicKeyStart, "test-group", @@ -703,7 +705,7 @@ func TestConsumer_NackLifecycleMetrics(t *testing.T) { reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") - c := New(logger, scope, reg, errs.NewClassifierProcessor()) + c := New(logger, scope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) handler := &testController{} setupController(handler, "test-handler", testTopicKeyStart, "test-group", @@ -758,7 +760,7 @@ func TestConsumer_PerPartitionProcessing(t *testing.T) { reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) // Track processing by partition partBDone := make(chan struct{}) @@ -843,7 +845,7 @@ func TestConsumer_PartitionOrdering(t *testing.T) { reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := 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. @@ -912,7 +914,7 @@ func TestConsumer_PartitionWorkerCleanup(t *testing.T) { reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) processedCount := int64(0) @@ -964,7 +966,7 @@ func TestConsumer_ConsumeLoopSurvivesCallerDeadline(t *testing.T) { reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") - c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor()) + c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New()) processed := make(chan string, 1) handler := &testController{} @@ -999,3 +1001,314 @@ 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 + changed chan struct{} + err error + + parked chan consumergate.Parked + released chan string // message IDs +} + +func newFakeGate() *fakeGate { + return &fakeGate{ + closed: make(map[consumergate.Key]bool), + changed: make(chan struct{}), + parked: make(chan consumergate.Parked, 16), + released: make(chan string, 16), + } +} + +func (f *fakeGate) close(key consumergate.Key) { + f.mu.Lock() + defer f.mu.Unlock() + if f.closed[key] { + return + } + f.closed[key] = true + f.signalChanged() +} + +func (f *fakeGate) open(key consumergate.Key) { + f.mu.Lock() + defer f.mu.Unlock() + if !f.closed[key] { + return + } + delete(f.closed, key) + f.signalChanged() +} + +func (f *fakeGate) signalChanged() { + close(f.changed) + f.changed = make(chan struct{}) +} + +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 Watch mimics the contract: it stamps the entered +// identity on the parked descriptor, announces it on the parked channel, and a +// monitor goroutine waits for gate-state change signals until the gate opens or +// ctx is cancelled; on open it sends the message ID on the released channel and +// yields nil on the watch 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) Watch(context.Context, consumergate.DeliveryDescriptor) <-chan error { + ch := make(chan error, 1) + ch <- nil + return ch +} + +// 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) Watch(ctx context.Context, descriptor consumergate.DeliveryDescriptor) <-chan error { + parked := consumergate.Parked{ + ConsumerGroup: e.key.ConsumerGroup, + Topic: descriptor.Topic, + MessageID: descriptor.MessageID, + PartitionKey: e.key.PartitionKey, + Payload: descriptor.Payload, + Attempt: descriptor.Attempt, + } + // Record synchronously so the parked descriptor is observable by the time + // Watch returns, mirroring the file store's synchronous recordParked. + e.gate.parked <- parked + + ch := make(chan error, 1) + go func() { + for { + e.gate.mu.Lock() + closed := e.gate.isClosed(e.key.ConsumerGroup, e.key.PartitionKey) + changed := e.gate.changed + e.gate.mu.Unlock() + if !closed { + e.gate.released <- parked.MessageID + ch <- nil + return + } + + select { + case <-ctx.Done(): + ch <- ctx.Err() + return + case <-changed: + } + } + }() + return ch +} + +// 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, Delivery) error) (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("start"), "test-group") + + c := New(zaptest.NewLogger(t).Sugar(), tally.NoopScope, reg, errs.NewClassifierProcessor(), gate) + + handler := &testController{} + setupController(handler, "test-handler", TopicKey("start"), "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 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.close(consumergate.Key{ConsumerGroup: "test-group"}) + + var processed atomic.Bool + c, deliveryChan := startGatedConsumer(t, ctrl, gate, func(context.Context, 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("start").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.open(consumergate.Key{ConsumerGroup: "test-group"}) + 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.close(consumergate.Key{ConsumerGroup: "test-group", PartitionKey: "gated-partition"}) + + var handled sync.Map + c, deliveryChan := startGatedConsumer(t, ctrl, gate, func(_ context.Context, delivery 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.open(consumergate.Key{ConsumerGroup: "test-group", PartitionKey: "gated-partition"}) + <-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.close(consumergate.Key{ConsumerGroup: "test-group"}) + + var processed atomic.Bool + c, deliveryChan := startGatedConsumer(t, ctrl, gate, func(context.Context, 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.close(consumergate.Key{ConsumerGroup: "test-group"}) + gate.setErr(fmt.Errorf("gate medium unavailable")) + + handledMsg := "" + c, deliveryChan := startGatedConsumer(t, ctrl, gate, func(_ context.Context, delivery 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..244ed9d9 --- /dev/null +++ b/platform/extension/consumergate/README.md @@ -0,0 +1,25 @@ +# Consumer Gate Extension + +Runtime stop/start of individual queue controllers without stopping the service that hosts them. The file implementation is explicitly scoped to deterministic e2e scenarios and single-host development; fleet-wide production pause requires a shared backend that is not part of this change. 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 `Watch` records the parked delivery (stamping the entered identity and `ParkedAtMs`) and returns a channel that yields once — `nil` when the gate opens, or an error if gate state cannot be read or written. The record is removed before `Watch` yields on every terminal path, so parked records describe only deliveries currently blocked behind a gate. Handing back a channel rather than blocking lets the caller multiplex the wait against its own events (context cancellation, visibility extension) in a single select; the package-level `Wait` helper wraps `Watch` for callers that only need the simple blocking behaviour. Stopping is a barrier, not preemption — a delivery already past its gate is not recalled. + +The gate does not own the source queue delivery and cannot acknowledge, nack, reject, remove, or move it. The caller remains responsible for queue lifecycle while a blocked entry is watched. + +The package defines three interfaces, the package-level `Wait` helper, 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, watches with `Watch`, supplying a `DeliveryDescriptor` containing only caller-owned message data. The implementation combines that descriptor with the gate identity captured by `Enter` and its own parked timestamp to create the observable `Parked` record. `Watch` returns a channel; the free function `Wait` blocks on it for callers that do not multiplex. 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). Once the wait ends, the record is removed so the parked tree remains a view of current state rather than an unbounded delivery history. + +## Failure posture + +An `Enter` or `Watch` 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/) — an explicit E2E and single-host development implementation using a shared directory. 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. It coordinates only processes that see that directory and is not a fleet-wide production control plane. +- [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..0a646c44 --- /dev/null +++ b/platform/extension/consumergate/consumergate.go @@ -0,0 +1,201 @@ +// 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 holds it. A blocked Entry does not +// block the caller — Entry.Watch records the parked delivery and returns a +// channel that yields exactly one result: nil when the gate opens, or an error +// if gate state cannot be read or the record written. This lets the caller +// multiplex the wait against its own events (context cancellation, visibility +// extension) in a single select. Callers that only need the simple blocking +// behaviour use the package-level Wait helper. The gate owns the monitoring +// goroutine and the parked-delivery observation records: it records the parked +// delivery before monitoring (stamping ParkedAtMs) and removes the record when +// monitoring ends, so parked records describe only deliveries currently held +// behind a gate. +// +// The package holds the contract only: Gate and Entry (the admission +// interfaces), the Wait helper, 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 +} + +// DeliveryDescriptor is the caller-owned description of a delivery that may +// be parked. It contains only values known by the consumer; the gate +// implementation owns the gate identity and parked timestamp added to the +// observable Parked record. +type DeliveryDescriptor struct { + // 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 + + // 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 +} + +// Parked is the gate-owned observation record for one blocked delivery. +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. + ParkedAtMs 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 its Watch monitors the gate until it + // opens. + // + // 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 outcome of Gate.Enter for one delivery. +type Entry interface { + // Blocked reports whether the gate was closed when the delivery entered. + // An unblocked entry needs no Watch — the delivery may proceed at once. + Blocked() bool + + // Watch records delivery as parked, adding the gate identity captured by + // Enter and the implementation-owned ParkedAtMs, and begins monitoring the + // gate. It returns a channel that yields exactly one value: nil when the gate + // opens, or a non-nil error if gate state could not be read or the record + // written — without further interpretation, as what to do with a failed wait + // is the caller's policy. If ctx is cancelled while monitoring, the channel + // yields ctx.Err(). The implementation removes the parked record before + // yielding on every terminal path, so ListParked contains only deliveries + // currently blocked behind a gate. + // + // Watch observes only the descriptor. It does not own or mutate the source + // queue delivery: acknowledging, nacking, rejecting, removing, or moving the + // delivery remains the caller's responsibility. + // + // The returned channel is buffered so the monitoring goroutine never blocks + // on its single send: a caller may cancel ctx and walk away without draining + // it, and the goroutine still exits. Watch must be called at most once per + // blocked Entry. + Watch(ctx context.Context, descriptor DeliveryDescriptor) <-chan error +} + +// Wait blocks until the gate behind entry opens or fails, or ctx is cancelled. +// It is the simple blocking adapter over Entry.Watch for callers (and tests) +// that do not need to multiplex the wait against other events: it returns nil +// for an unblocked entry or when the gate opens, the gate's error if the wait +// failed, or ctx.Err() after the watcher observes cancellation and completes +// its cleanup. +func Wait(ctx context.Context, entry Entry, descriptor DeliveryDescriptor) error { + if !entry.Blocked() { + return nil + } + return <-entry.Watch(ctx, descriptor) +} + +// 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 delivery currently parked for the consumer group. + // Callers may filter by topic or message ID. + 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..14646973 --- /dev/null +++ b/platform/extension/consumergate/file/README.md @@ -0,0 +1,23 @@ +# File-Backed Consumer Gate + +Stores gate state as plain files under an explicitly configured root directory for E2E tests and single-host development. 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, and `parked_at_ms` while a delivery is blocked. The record is deleted before the wait ends, so the parked tree contains only active waits and does not retain payloads after release or cancellation. All writes go through temp-file-plus-rename so readers never see partial JSON. + +Gate state is not cached. Each delivery reads its applicable gate files, and each blocked delivery polls those files at the configured interval until they are absent. + +## 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 E2E stack bind-mounts one test-owned host directory into every service container, and the test manipulates gates and reads parked records as local files. A file gates only processes that see that directory. State survives a process or container restart only if the configured directory survives it, and it does not survive node replacement without shared persistent storage. This implementation is not a fleet-wide production control plane; a production pause feature requires another shared `consumergate` backend. diff --git a/platform/extension/consumergate/file/store.go b/platform/extension/consumergate/file/store.go new file mode 100644 index 00000000..ac060945 --- /dev/null +++ b/platform/extension/consumergate/file/store.go @@ -0,0 +1,388 @@ +// 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 reads the applicable gate files for every delivery. A blocked Entry's +// Watch writes the parked record, then a monitor goroutine polls those files on +// a ticker at the configured interval. The parked record is removed before the +// watch yields on every terminal path, so the directory contains only +// deliveries currently held behind a gate. +// +// Filesystem events such as inotify are intentionally not the correctness +// mechanism here. They are platform-specific, can overflow or coalesce events, +// and require watches to be re-established when watched paths are removed or +// replaced. Event behavior also varies across bind mounts, overlay or network +// filesystems, rootless Docker, and Docker Desktop's host/container filesystem +// bridge. Polling works consistently across those environments. A future +// enhancement may use filesystem events to accelerate wakeups while retaining +// polling as the fallback and convergence mechanism. +// +// The medium is deliberately scoped to E2E tests and single-host development: +// pausing a controller is writing a small file, resuming is rm, and a bind mount +// makes the state reachable outside the service process. A file gates only +// processes that see the directory; it is not a fleet-wide production control +// plane. +package file + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/url" + "os" + "path/filepath" + "strings" + "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 +} + +// 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, + } +} + +// 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 +} + +// 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.isGated(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 } + +// Watch implements consumergate.Entry. An open gate never blocks and records +// nothing; the returned channel yields nil at once. +func (openEntry) Watch(context.Context, consumergate.DeliveryDescriptor) <-chan error { + ch := make(chan error, 1) + ch <- nil + return ch +} + +// 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 } + +// Watch implements consumergate.Entry. It records the parked delivery (stamping +// the entry's identity and ParkedAtMs) synchronously, then spawns a goroutine +// that polls on a ticker at the store's poll interval and yields exactly one +// value on the returned channel: nil when the gate opens, the read/write error +// if gate state cannot be read or the record written, or ctx.Err() if ctx is +// cancelled first. The parked record is removed before any result is yielded. +// The channel is buffered so the goroutine never blocks on its send. +func (e *parkedEntry) Watch(ctx context.Context, descriptor consumergate.DeliveryDescriptor) <-chan error { + s := e.store + ch := make(chan error, 1) + + // Construct the gate-owned observation from the caller's delivery + // description and the identity captured by Enter. Record synchronously so + // the parked record exists by the time Watch returns. + parked := consumergate.Parked{ + ConsumerGroup: e.key.ConsumerGroup, + Topic: descriptor.Topic, + MessageID: descriptor.MessageID, + PartitionKey: e.key.PartitionKey, + Payload: descriptor.Payload, + Attempt: descriptor.Attempt, + ParkedAtMs: time.Now().UnixMilli(), + } + if err := s.recordParked(parked); err != nil { + ch <- err + return ch + } + + go func() { + finish := func(waitErr error) { + removeErr := s.removeParked(parked.ConsumerGroup, parked.Topic, parked.MessageID) + ch <- errors.Join(waitErr, removeErr) + } + + ticker := time.NewTicker(s.pollInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + finish(ctx.Err()) + return + case <-ticker.C: + } + + gated, err := s.isGated(e.key.ConsumerGroup, e.key.PartitionKey) + if err != nil { + finish(err) + return + } + if !gated { + finish(nil) + return + } + } + }() + + return ch +} + +// 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 +} + +// removeParked removes a parked-delivery record. Removing an already-absent +// record is a no-op. +func (s *Store) removeParked(consumerGroup, topic, messageID string) error { + path := s.parkedPath(consumerGroup, topic, messageID) + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("failed to remove 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"` +} + +// 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..32ff8288 --- /dev/null +++ b/platform/extension/consumergate/file/store_test.go @@ -0,0 +1,331 @@ +// 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" + "time" + + "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} + +// awaitParked indefinitely waits for a parked record to appear in the store, returning the records. +// It will wait up until the test times out. +func awaitParked(t *testing.T, store *Store, ctx context.Context, consumerGroup string) []consumergate.Parked { + t.Helper() + ticker := time.NewTicker(time.Duration(testCfg.PollIntervalMs) * time.Millisecond) + defer ticker.Stop() + for { + records, err := store.ListParked(ctx, consumerGroup) + require.NoError(t, err) + if len(records) > 0 { + return records + } + <-ticker.C + } +} + +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]) + + // 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.removeParked(parked.ConsumerGroup, parked.Topic, parked.MessageID)) + records, err = store.ListParked(ctx, parked.ConsumerGroup) + require.NoError(t, err) + assert.Empty(t, records) + + // Removing an already-absent record is a no-op. + require.NoError(t, store.removeParked(parked.ConsumerGroup, parked.Topic, parked.MessageID)) +} + +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, consumergate.Wait(ctx, entry, consumergate.DeliveryDescriptor{ + 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 <- consumergate.Wait(ctx, entry, consumergate.DeliveryDescriptor{ + Topic: "topic", + MessageID: "msg-1", + Payload: []byte("hello"), + Attempt: 1, + }) + }() + + records := awaitParked(t, store, ctx, "group") + 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 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 remove the active parked + // record before returning. + require.NoError(t, store.Open(ctx, key)) + require.NoError(t, <-waitDone) + + records, err = store.ListParked(ctx, "group") + require.NoError(t, err) + assert.Empty(t, records) +} + +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 <- consumergate.Wait(ctx, entry, consumergate.DeliveryDescriptor{ + Topic: "topic", + MessageID: "msg-1", + Payload: []byte("hello"), + Attempt: 1, + }) + }() + + // Wait until the parked record appears, then cancel. + awaitParked(t, store, ctx, "group") + + cancel() + require.ErrorIs(t, <-waitDone, context.Canceled) + + // Cancellation ends the active wait, so its parked record is removed. + records, err := store.ListParked(context.Background(), "group") + require.NoError(t, err) + assert.Empty(t, records) +} + +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..c956d9d0 --- /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)) +} + +// Watch mocks base method. +func (m *MockEntry) Watch(ctx context.Context, descriptor consumergate.DeliveryDescriptor) <-chan error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Watch", ctx, descriptor) + ret0, _ := ret[0].(<-chan error) + return ret0 +} + +// Watch indicates an expected call of Watch. +func (mr *MockEntryMockRecorder) Watch(ctx, descriptor any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Watch", reflect.TypeOf((*MockEntry)(nil).Watch), ctx, descriptor) +} + +// 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..fa4c37a0 --- /dev/null +++ b/platform/extension/consumergate/noop/gate.go @@ -0,0 +1,55 @@ +// 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 } + +// Watch implements consumergate.Entry. A no-op gate never blocks, so the +// returned channel yields nil at once. It is never reached in practice because +// Blocked reports false. +func (Gate) Watch(context.Context, consumergate.DeliveryDescriptor) <-chan error { + ch := make(chan error, 1) + ch <- nil + return ch +} diff --git a/platform/extension/consumergate/noop/gate_test.go b/platform/extension/consumergate/noop/gate_test.go new file mode 100644 index 00000000..ae5ea67c --- /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, consumergate.Wait(context.Background(), entry, consumergate.DeliveryDescriptor{ + Topic: "topic", + MessageID: "msg-1", + Payload: []byte("hello"), + Attempt: 1, + })) +} diff --git a/service/runway/server/BUILD.bazel b/service/runway/server/BUILD.bazel index fec51e28..12e4af7c 100644 --- a/service/runway/server/BUILD.bazel +++ b/service/runway/server/BUILD.bazel @@ -20,6 +20,9 @@ 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/consumergate/noop: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/Dockerfile b/service/runway/server/Dockerfile index e70ebd77..04f01cee 100644 --- a/service/runway/server/Dockerfile +++ b/service/runway/server/Dockerfile @@ -1,11 +1,12 @@ FROM debian:bookworm-slim -RUN apt-get update && apt-get install -y ca-certificates && rm -rf /var/lib/apt/lists/* -WORKDIR /root/ +RUN apt-get update && apt-get install -y ca-certificates && rm -rf /var/lib/apt/lists/* \ + && mkdir -p /app && chmod 0755 /app +WORKDIR /app # Built via: make build-runway-linux -COPY .docker-bin/runway ./runway +COPY --chmod=0555 .docker-bin/runway ./runway EXPOSE 8080 -CMD ["./runway"] +CMD ["/app/runway"] diff --git a/service/runway/server/main.go b/service/runway/server/main.go index ee6a0936..00deacf7 100644 --- a/service/runway/server/main.go +++ b/service/runway/server/main.go @@ -34,6 +34,9 @@ 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" + 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/runway/controller" @@ -152,6 +155,7 @@ func run() error { genericerrs.Classifier, mysqlerrs.Classifier, ), + newConsumerGate(logger), ) mergerFactory := newMergerFactory() @@ -288,3 +292,17 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe }, }) } + +// newConsumerGate enables the file-backed consumer gate only when +// CONSUMER_GATE_DIR is explicitly configured. The file implementation is for +// E2E and single-host development; normal service deployments use the no-op +// implementation. +func newConsumerGate(logger *zap.Logger) consumergate.Gate { + dir := os.Getenv("CONSUMER_GATE_DIR") + if dir == "" { + logger.Info("consumer gate disabled") + return consumergatenoop.New() + } + 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 ddc13418..d82bbcbe 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/service/stovepipe/server/BUILD.bazel @@ -16,6 +16,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 8df878af..062c14b7 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" @@ -228,15 +229,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(), ) // Each factory is constructed once and threaded through every consumer of diff --git a/service/submitqueue/docker-compose.yml b/service/submitqueue/docker-compose.yml index 2b11dc2f..d3dd0b56 100644 --- a/service/submitqueue/docker-compose.yml +++ b/service/submitqueue/docker-compose.yml @@ -7,6 +7,16 @@ # # Quick start: # make e2e-test +# +# Consumer gate: every service shares one host directory (bind-mounted at +# /var/submitqueue/consumergate) so E2E tests and local developers 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. +# SQ_CONTAINER_USER controls the UID:GID of the application services that write +# parked records into that bind mount. It defaults to root for normal local +# usage; the e2e suite selects the host user for rootful Docker and root for +# rootless Docker, where container root maps to the host user. services: # Application Database - Stores business data (requests, counters, etc.) @@ -54,6 +64,7 @@ services: build: context: ${REPO_ROOT} dockerfile: service/submitqueue/gateway/server/Dockerfile + user: "${SQ_CONTAINER_USER:-0:0}" ports: - "8080" # Random ephemeral port to avoid conflicts environment: @@ -63,9 +74,13 @@ services: # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true # Path to YAML queue configuration baked into the image - - QUEUE_CONFIG_PATH=/root/queues.yaml + - QUEUE_CONFIG_PATH=/app/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 @@ -77,6 +92,7 @@ services: build: context: ${REPO_ROOT} dockerfile: service/submitqueue/orchestrator/server/Dockerfile + user: "${SQ_CONTAINER_USER:-0:0}" ports: - "8080" # Random ephemeral port to avoid conflicts environment: @@ -86,6 +102,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 @@ -102,6 +122,7 @@ services: build: context: ${REPO_ROOT} dockerfile: service/runway/server/Dockerfile + user: "${SQ_CONTAINER_USER:-0:0}" ports: - "8080" # Random ephemeral port to avoid conflicts environment: @@ -109,6 +130,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 711d1279..b4030956 100644 --- a/service/submitqueue/gateway/server/BUILD.bazel +++ b/service/submitqueue/gateway/server/BUILD.bazel @@ -20,6 +20,9 @@ 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/consumergate/noop: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/Dockerfile b/service/submitqueue/gateway/server/Dockerfile index 73840ae0..b9486b4c 100644 --- a/service/submitqueue/gateway/server/Dockerfile +++ b/service/submitqueue/gateway/server/Dockerfile @@ -1,16 +1,17 @@ FROM debian:bookworm-slim -RUN apt-get update && apt-get install -y ca-certificates && rm -rf /var/lib/apt/lists/* -WORKDIR /root/ +RUN apt-get update && apt-get install -y ca-certificates && rm -rf /var/lib/apt/lists/* \ + && mkdir -p /app && chmod 0755 /app +WORKDIR /app # Copy pre-built Linux binary # Built via: make build-gateway-linux -COPY .docker-bin/gateway ./gateway +COPY --chmod=0555 .docker-bin/gateway ./gateway # Sample queue configuration; the gateway reads it on startup via # QUEUE_CONFIG_PATH (set in docker-compose.yml). -COPY service/submitqueue/gateway/server/queues.yaml ./queues.yaml +COPY --chmod=0444 service/submitqueue/gateway/server/queues.yaml ./queues.yaml EXPOSE 8080 -CMD ["./gateway"] +CMD ["/app/gateway"] diff --git a/service/submitqueue/gateway/server/docker-compose.yml b/service/submitqueue/gateway/server/docker-compose.yml index 7d439b60..a896f2c6 100644 --- a/service/submitqueue/gateway/server/docker-compose.yml +++ b/service/submitqueue/gateway/server/docker-compose.yml @@ -63,7 +63,7 @@ services: # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true # Path to YAML queue configuration baked into the image - - QUEUE_CONFIG_PATH=/root/queues.yaml + - QUEUE_CONFIG_PATH=/app/queues.yaml # Stable subscriber name for the request-log consumer - HOSTNAME=gateway-dev depends_on: diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index 7a46023c..6cd20701 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -33,6 +33,9 @@ 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" + consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop" 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 +364,7 @@ func run() error { genericerrs.Classifier, mysqlerrs.Classifier, ), + newConsumerGate(logger), ) logController := logctrl.NewController(logger.Sugar(), scope, store, topickey.TopicKeyLog, "gateway-log") @@ -439,3 +443,17 @@ func run() error { return err } + +// newConsumerGate enables the file-backed consumer gate only when +// CONSUMER_GATE_DIR is explicitly configured. The file implementation is for +// E2E and single-host development; normal service deployments use the no-op +// implementation. +func newConsumerGate(logger *zap.Logger) consumergate.Gate { + dir := os.Getenv("CONSUMER_GATE_DIR") + if dir == "" { + logger.Info("consumer gate disabled") + return consumergatenoop.New() + } + 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 0b64c2e7..02d15784 100644 --- a/service/submitqueue/orchestrator/server/BUILD.bazel +++ b/service/submitqueue/orchestrator/server/BUILD.bazel @@ -20,6 +20,9 @@ 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/consumergate/noop: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/Dockerfile b/service/submitqueue/orchestrator/server/Dockerfile index fb1a4626..ff844e9d 100644 --- a/service/submitqueue/orchestrator/server/Dockerfile +++ b/service/submitqueue/orchestrator/server/Dockerfile @@ -1,12 +1,13 @@ FROM debian:bookworm-slim -RUN apt-get update && apt-get install -y ca-certificates && rm -rf /var/lib/apt/lists/* -WORKDIR /root/ +RUN apt-get update && apt-get install -y ca-certificates && rm -rf /var/lib/apt/lists/* \ + && mkdir -p /app && chmod 0755 /app +WORKDIR /app # Copy pre-built Linux binary # Built via: make build-orchestrator-linux -COPY .docker-bin/orchestrator ./orchestrator +COPY --chmod=0555 .docker-bin/orchestrator ./orchestrator EXPOSE 8080 -CMD ["./orchestrator"] +CMD ["/app/orchestrator"] diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index eada3e64..86afe67b 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -37,6 +37,9 @@ 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" + consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop" "github.com/uber/submitqueue/platform/extension/counter" mysqlcounter "github.com/uber/submitqueue/platform/extension/counter/mysql" extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" @@ -215,6 +218,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, @@ -223,9 +231,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 @@ -717,6 +727,20 @@ func getEnv(key, defaultVal string) string { return defaultVal } +// newConsumerGate enables the file-backed consumer gate only when +// CONSUMER_GATE_DIR is explicitly configured. The file implementation is for +// E2E and single-host development; normal service deployments use the no-op +// implementation. +func newConsumerGate(logger *zap.Logger) consumergate.Gate { + dir := os.Getenv("CONSUMER_GATE_DIR") + if dir == "" { + logger.Info("consumer gate disabled") + return consumergatenoop.New() + } + 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/test/e2e/stovepipe/harness_test.go b/test/e2e/stovepipe/harness_test.go index 6aa812c9..8eb517c5 100644 --- a/test/e2e/stovepipe/harness_test.go +++ b/test/e2e/stovepipe/harness_test.go @@ -24,17 +24,29 @@ package e2e_test // - the asynchronous completion of the process stage by polling the queue // backend's per-consumer-group delivery state until the message is acked. // -// Convergence is bounded by require.Eventually rather than time.Sleep: the -// process consumer runs inside the stovepipe-service container, so there is no -// in-process signal to await; a timeout here means the stage is genuinely stuck, -// not a timing race. +// The process consumer runs inside the stovepipe-service container, so there is +// no in-process signal to await. Polling continues until the condition holds or +// Bazel's test timeout terminates a genuinely stuck suite. import ( + "time" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" pb "github.com/uber/submitqueue/api/stovepipe/protopb" ) +func pollUntil(interval time.Duration, condition func() bool) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + if condition() { + return + } + <-ticker.C + } +} + // The process consumer's topic and consumer group as wired in // service/stovepipe/server/main.go (topic name "process", consumer group // "stovepipe-process"). awaitProcessed reads the queue backend's delivery state @@ -92,12 +104,11 @@ func (s *StovepipeE2ESuite) publishedMessageCount(id string) int { // as the message's partition key (see the ingest controller), so the partition // key here is the queue. func (s *StovepipeE2ESuite) awaitProcessed(queue string) { - t := s.T() const query = ` - SELECT offset_acked - FROM queue_offsets - WHERE consumer_group = ? AND topic = ? AND partition_key = ?` - require.Eventually(t, func() bool { + SELECT offset_acked + FROM queue_offsets + WHERE consumer_group = ? AND topic = ? AND partition_key = ?` + pollUntil(processPollInterval, func() bool { var ackedOffset int64 err := s.queueDB.QueryRow(query, processConsumerGroup, processTopic, queue).Scan(&ackedOffset) if err != nil { @@ -107,9 +118,7 @@ func (s *StovepipeE2ESuite) awaitProcessed(queue string) { } s.log.Logf("acked offset for queue %s = %d (want > 0)", queue, ackedOffset) return ackedOffset > 0 - }, processTimeout, processPollInterval, - "process consumer group %q on topic %q should advance the acked offset for queue %s", - processConsumerGroup, processTopic, queue) + }) } // assertIngestPersisted asserts the synchronous side effects of a successful diff --git a/test/e2e/stovepipe/suite_test.go b/test/e2e/stovepipe/suite_test.go index 383a1432..365cb808 100644 --- a/test/e2e/stovepipe/suite_test.go +++ b/test/e2e/stovepipe/suite_test.go @@ -54,12 +54,9 @@ import ( // suite can only observe its completion black-box through the queue backend's // delivery-state table — there is no in-process signal to await across the // container boundary. A bounded poll is the deterministic-enough analog: -// processTimeout is a safety net (a failure here means the stage is genuinely -// stuck, not a timing race) and processPollInterval bounds re-query frequency. -const ( - processTimeout = 30 * time.Second - processPollInterval = 500 * time.Millisecond -) +// processPollInterval bounds re-query frequency; Bazel's test timeout is the +// only convergence deadline. +const processPollInterval = 500 * time.Millisecond type StovepipeE2ESuite struct { suite.Suite diff --git a/test/e2e/submitqueue/BUILD.bazel b/test/e2e/submitqueue/BUILD.bazel index da0f7b0f..8ba86510 100644 --- a/test/e2e/submitqueue/BUILD.bazel +++ b/test/e2e/submitqueue/BUILD.bazel @@ -27,8 +27,11 @@ go_test( 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..ad54fdc2 100644 --- a/test/e2e/submitqueue/harness_test.go +++ b/test/e2e/submitqueue/harness_test.go @@ -21,22 +21,34 @@ package e2e_test // - black-box, by polling the GetRequestSummaryByID RPC to a target/terminal status; and // - black-box, by reading the ordered stage progression through GetRequestHistoryByID. // -// Convergence is bounded by require.Eventually (persistTimeout / -// persistPollInterval) rather than time.Sleep: the pipeline consumers run inside -// containers, so there is no in-process signal to await; a timeout here means a -// stage is genuinely stuck, not a timing race. +// The pipeline consumers run inside containers, so there is no in-process +// signal to await. Polling continues until the condition holds or Bazel's test +// timeout terminates a genuinely stuck suite. 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" ) +func pollUntil(interval time.Duration, condition func() bool) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + if condition() { + return + } + <-ticker.C + } +} + // land submits a request with the default REBASE strategy and returns its sqid. // URIs may carry "sq-fake=" markers to steer negative paths (see // submitqueue/core/fakemarker); the happy path uses a plain change URI. @@ -67,8 +79,7 @@ func (s *E2EIntegrationSuite) currentStatus(sqid string) (entity.RequestStatus, // awaitStatus polls GetRequestSummaryByID until the request reaches exactly want. func (s *E2EIntegrationSuite) awaitStatus(sqid string, want entity.RequestStatus) { - t := s.T() - require.Eventually(t, func() bool { + pollUntil(persistPollInterval, func() bool { got, err := s.currentStatus(sqid) if err != nil { s.log.Logf("GetRequestSummaryByID(%s) not ready yet: %v", sqid, err) @@ -76,16 +87,14 @@ func (s *E2EIntegrationSuite) awaitStatus(sqid string, want entity.RequestStatus } s.log.Logf("GetRequestSummaryByID(%s) = %q (want %q)", sqid, got, want) return got == want - }, persistTimeout, persistPollInterval, - "request %s should reach status %q", sqid, want) + }) } // awaitTerminal polls GetRequestSummaryByID until the request reaches a terminal status // (landed, error, or cancelled) and returns it. func (s *E2EIntegrationSuite) awaitTerminal(sqid string) entity.RequestStatus { - t := s.T() var last entity.RequestStatus - require.Eventually(t, func() bool { + pollUntil(persistPollInterval, func() bool { got, err := s.currentStatus(sqid) if err != nil { s.log.Logf("GetRequestSummaryByID(%s) not ready yet: %v", sqid, err) @@ -94,8 +103,7 @@ func (s *E2EIntegrationSuite) awaitTerminal(sqid string) entity.RequestStatus { last = got s.log.Logf("GetRequestSummaryByID(%s) = %q (awaiting terminal)", sqid, got) return isTerminalStatus(got) - }, persistTimeout, persistPollInterval, - "request %s should reach a terminal status", sqid) + }) return last } @@ -130,6 +138,81 @@ 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 { + t := s.T() + var found consumergate.Parked + pollUntil(persistPollInterval, func() bool { + records, err := s.gate.ListParked(s.ctx, consumerGroup) + require.NoError(t, err, "failed to list parked deliveries for gate %s", consumerGroup) + for _, r := range records { + if r.Topic == topic && r.MessageID == messageID { + found = r + return true + } + } + return false + }) + return found +} + +// awaitUnparked polls until the previously observed parked record is absent. +// The gate removes the record before releasing the delivery, so disappearance +// proves the delivery cleared the gate after it opened. +func (s *E2EIntegrationSuite) awaitUnparked(consumerGroup, topic, messageID string) { + t := s.T() + pollUntil(persistPollInterval, func() bool { + records, err := s.gate.ListParked(s.ctx, consumerGroup) + require.NoError(t, err, "failed to list parked deliveries for gate %s", consumerGroup) + for _, r := range records { + if r.Topic == topic && r.MessageID == messageID { + return false + } + } + return true + }) +} + // 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 d4085da3..478bdc68 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/test/e2e/submitqueue/suite_test.go @@ -27,6 +27,10 @@ package e2e_test import ( "context" "database/sql" + "fmt" + "os" + "os/exec" + "strings" "testing" "time" @@ -34,8 +38,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" @@ -52,9 +59,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) { @@ -62,13 +70,10 @@ func TestE2EIntegration(t *testing.T) { } // The gateway log consumer runs inside the gateway-service container, so there -// is no in-process signal to wait on across the container boundary. A bounded -// GetRequestSummaryByID poll is therefore the deterministic-enough analog: persistTimeout -// is a safety net, and persistPollInterval bounds how often we re-query. -const ( - persistTimeout = 30 * time.Second - persistPollInterval = 500 * time.Millisecond -) +// is no in-process signal to wait on across the container boundary. +// persistPollInterval bounds how often helpers re-query; Bazel's test timeout is +// the only convergence deadline. +const persistPollInterval = 500 * time.Millisecond func (s *E2EIntegrationSuite) SetupSuite() { t := s.T() @@ -77,6 +82,21 @@ func (s *E2EIntegrationSuite) SetupSuite() { s.log.Logf("Starting E2E integration test suite using docker-compose") + // Application services write parked records into a host bind mount. On a + // rootful daemon they must run as the host test user so those records remain + // readable and removable by the test. On a rootless daemon, container root + // already maps to the host user, so keep the container user at 0:0. + containerUser := dockerContainerUser(t) + t.Setenv("SQ_CONTAINER_USER", containerUser) + s.log.Logf("Application containers will run as %s", containerUser) + + // Consumer-gate state is an explicit E2E-only opt-in. The compose file + // bind-mounts this test-owned directory into every application service, and + // the suite manipulates the same directory through the file implementation. + gateDir := t.TempDir() + t.Setenv("SQ_CONSUMER_GATE_DIR", gateDir) + s.gate = consumergatefile.New(gateDir, consumergate.DefaultConfig()) + // Use docker-compose from service/submitqueue (full stack), resolved from // the test runfiles. All three service images are built from a staged // build context assembled entirely from declared data dependencies. @@ -133,6 +153,21 @@ func (s *E2EIntegrationSuite) SetupSuite() { s.log.Logf("E2E integration test suite ready") } +// dockerContainerUser returns the UID:GID that application containers should +// use for host-bind-mounted test artifacts. Rootless Docker maps container root +// to the host user; rootful Docker needs the host UID:GID explicitly. +func dockerContainerUser(t *testing.T) string { + t.Helper() + + cmd := exec.Command("docker", "info", "--format", "{{json .SecurityOptions}}") + output, err := cmd.Output() + require.NoError(t, err, "failed to inspect Docker security options") + if strings.Contains(string(output), "name=rootless") { + return "0:0" + } + return fmt.Sprintf("%d:%d", os.Getuid(), os.Getgid()) +} + func (s *E2EIntegrationSuite) TearDownSuite() { t := s.T() s.log.Logf("Tearing down E2E integration test suite") @@ -306,32 +341,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. // -// 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() { +// 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. +// +// 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 cleared the gate. + s.openGate(gateGroup, queue) + s.awaitUnparked(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 3d9c3a48..b7ec937f 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() {