From 76daee309943ef13493d8247eef75681a836f208 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Sat, 1 Aug 2026 11:11:57 -0700 Subject: [PATCH] feat(consumergate)!: rebuild the gate on the postpone primitive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The gate parked blocked deliveries in memory: the partition goroutine sat in a watch/extend loop, renewing the parked delivery's visibility until the gate opened. That mechanism only babysat the one delivery it parked — messages already fetched into the per-partition buffer behind it (up to BatchSize) had their visibility lapse, redelivered as duplicates that stuffed the buffer until the topic's routing loop stalled, and burned a retry attempt per lapse until they were spuriously dead-lettered without ever failing. It also held a goroutine and a live lease per blocked partition for the whole close. ### What? The pre-Process gate check now postpones instead of parking: a blocked delivery gets its parked record written (observability) and is postponed for a re-check delay (~1s) — back to the queue as a partition barrier, redelivering without retry cost, re-checking the gate on each redelivery. Every buffered delivery behind a closed gate is postponed in turn as the partition drains, so nothing waits in memory, nothing lapses, and Stop never has parked state to unwind. The admit path unconditionally removes the parked record, which is what releases e2e awaitUnparked. Contract: Entry.Watch and the Wait helper are replaced by Park/Unpark (record ops); Config/DefaultConfig and the never-implemented Factory are deleted — the file store no longer polls, since the re-check cadence is the consumer's postpone delay. Semantics changes: release is quantized to the re-check delay (same order as the old 1s file poll), and a released message is a fresh delivery (Attempt restarts at 1) rather than "the same attempt" — no test asserted either. Consumer-gate RFC, extension READMEs, and the consumer README are updated; in-memory parking moves to the RFC's Rejected list with the buffered-lapse hazard as the reason. ## Test Plan ✅ `make test` (83 targets) — reworked gate tests: blocked deliveries are parked + postponed with the re-check delay and never processed; redelivery after open unparks and processes; partition-scoped gating; Stop is immediate while gated; fail-open on read errors. ✅ `bazel test //test/e2e/submitqueue/...` — the cancel-caught-pre-batch scenario passes unchanged: awaitParked observes the record while closed, awaitUnparked observes its removal after open, the stale check is dropped, the sentinel lands. ✅ `make fmt`, `make mocks`. --- doc/rfc/consumer-gate.md | 23 +-- doc/rfc/consumer-hold.md | 2 +- doc/rfc/index.md | 2 +- platform/consumer/README.md | 4 +- platform/consumer/consumer.go | 144 +++++++-------- platform/consumer/consumer_test.go | 173 ++++++++---------- platform/extension/consumergate/README.md | 13 +- .../extension/consumergate/consumergate.go | 103 +++-------- .../extension/consumergate/file/README.md | 4 +- platform/extension/consumergate/file/store.go | 130 +++---------- .../extension/consumergate/file/store_test.go | 132 +++++-------- .../consumergate/mock/consumergate_mock.go | 67 +++---- platform/extension/consumergate/noop/gate.go | 14 +- .../extension/consumergate/noop/gate_test.go | 7 +- service/runway/server/main.go | 2 +- service/submitqueue/gateway/server/main.go | 2 +- .../submitqueue/orchestrator/server/main.go | 2 +- test/e2e/submitqueue/harness_test.go | 9 +- test/e2e/submitqueue/suite_test.go | 9 +- 19 files changed, 308 insertions(+), 534 deletions(-) diff --git a/doc/rfc/consumer-gate.md b/doc/rfc/consumer-gate.md index f290492d..342aac93 100644 --- a/doc/rfc/consumer-gate.md +++ b/doc/rfc/consumer-gate.md @@ -15,15 +15,15 @@ Nothing in the system expresses this today. The queue can be manipulated from ou ### The gate is consumer middleware, acting on deliveries before the controller -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 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 stopping one delivery stops exactly that partition and nothing else. And the framework **owns the delivery outcome** — controllers signal it only through `Process`'s return value — so a delivery can be stopped simply by never 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 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. +The gate is a check installed by the consumer ahead of 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 check writes the observable parked record and then **postpones** the delivery — the same hold/postpone primitive controllers use to wait (see [Consumer Hold](consumer-hold.md)) — so the message goes back to the queue invisible for a short re-check delay, acts as a barrier its partition waits behind, and redelivers without consuming retry budget. Each redelivery re-checks the gate: still closed re-parks and re-postpones; open removes the parked record and proceeds into the controller in partition order. Nothing waits in memory — no goroutine blocks, no visibility lease is renewed, and a process death while a delivery is parked loses nothing, because the parked state *is* queue state. 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. -The gate is an external stop lever — closed and opened from outside the controller, ended by an event. A controller that itself needs to wait (backing off for a budget slot, polling a slow status) should not reach for the gate; that is the hold outcome — see [Consumer Hold](consumer-hold.md). +The gate is an external stop lever — closed and opened from outside the controller. A controller that itself needs to wait (backing off for a budget slot, polling a slow status) should not reach for the gate; it holds its own delivery directly — see [Consumer Hold](consumer-hold.md). The gate is the same postpone mechanism applied *before* the controller by an *external* decision. -One bounded side effect is accepted and documented: the routing loop feeds each partition through a channel buffered at the subscription's batch size, so if messages keep arriving for a parked partition, the topic's routing loop eventually stalls once that buffer fills. For a fully closed gate this is moot (every partition parks anyway), and at test volumes it never triggers. +The re-check loop has a bounded cost: while a gate is closed, each blocked partition redelivers its head message once per re-check delay (~1s), re-reading gate state and rewriting one parked record per cycle. At test volumes this is noise, and it buys the property that matters: deliveries already fetched into the consumer's in-memory buffer behind a blocked one are each postponed in turn as the partition drains, so nothing sits in memory accumulating visibility lapses. ### Gate identity: consumer group, optionally narrowed to a partition @@ -31,7 +31,7 @@ Every controller subscribes with a unique consumer group (`orchestrator-batch`, ### Gate state is a separate extension -The consumer gate is a shared extension in its own right, not a feature of any queue backend. The contract lives at `platform/extension/consumergate/`: the behavioral interface the 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. +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 and the write surface tests and tooling use. `Park` 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, and `Unpark` removes the record on the admit path. 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. @@ -45,7 +45,7 @@ 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, 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. +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; each re-check of a still-closed gate refreshes the record, and the admit path removes it once the gate opens, so payloads are not retained after release. All writes go through temp-file-plus-rename so readers never see partial JSON. Files are the simplest medium for the E2E and single-host scope: @@ -53,23 +53,23 @@ Files are the simplest medium for the E2E and single-host scope: - **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. - **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 is platform-specific, watches can overflow or require re-registration, and event behavior varies across bind mounts, overlay or network filesystems, rootless Docker, and Docker Desktop's host/container filesystem bridge. Polling is the portable convergence mechanism; filesystem events may be added later as an optional wakeup optimization alongside it. +The store never watches the directory: filesystem notifications such as inotify are platform-specific, can overflow or require re-registration, and behave differently across bind mounts, overlay or network filesystems, rootless Docker, and Docker Desktop's host/container filesystem bridge. Instead, gate state is re-read on every delivery attempt — and a blocked delivery's re-check cadence is the consumer's postpone delay, which is the portable convergence mechanism regardless of medium. 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. ### 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. +The check reads the applicable gate state for every delivery. A blocked delivery is postponed for a short re-check delay (~1s) and re-checks on redelivery. Closing a gate therefore affects the next delivery check without waiting for a cache refresh; opening one releases blocked deliveries within one re-check delay plus the queue's own 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. 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. +Parking writes the parked record before the delivery is postponed. 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. While the gate stays closed the record is continuously refreshed by the re-check loop; the admit path removes it once the gate opens, so records are bounded by currently parked messages and the directory is empty whenever no delivery is blocked 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. +If gate state cannot be read (directory missing, I/O error), the check logs, increments an error counter, and lets deliveries through. A failed parked-record write is logged and the delivery is still postponed — the record is observability, never the mechanism. A failed postpone is abandoned like a failed acknowledge: the visibility timeout lapses into a normal redelivery, which re-checks the gate (that redelivery consumes one retry attempt, bounded per lapse). 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 @@ -79,13 +79,14 @@ The cancellation scenario, expressed as stop → observe → start: 2. It lands a request. The orchestrator runs it to the merge-conflict-check hand-off; runway's subscriber delivers the check message, and the gate parks it. 3. The test awaits the parked record — proof the controller is stopped *and* holding exactly this message. Runway itself is still running; its RPC surface and merge controller are untouched. 4. While stopped, the test observes and acts: it cancels the request, awaits the terminal `cancelled` status through the existing event plane, and asserts no batch ever enrolled the request. -5. The test opens the gate. Within a refresh tick the parked delivery proceeds into the controller as the same attempt; runway answers the now-stale check, and the test asserts the signal is dropped for the halted request. +5. The test opens the gate. Within a re-check tick the postponed delivery redelivers, clears the open gate, and proceeds into the controller as a fresh attempt (postponing resets retry accounting); runway answers the now-stale check, and the test asserts the signal is dropped for the halted request. The two-controller interleaving scenario is the same shape: close controller X's gate, drive both messages, await X's parked record and Y's downstream effect in the required order, open X's gate. ## 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. +- **Parking deliveries in memory** (block the partition goroutine on a blocked delivery, extending its visibility until the gate opens — this design's original mechanism). It holds a goroutine and a lease per blocked partition, and it only babysits the delivery it parks: messages already fetched into the per-partition buffer behind it (up to the batch size) have their visibility lapse, redeliver as duplicates that stuff the buffer until the topic's routing loop stalls, and burn a retry attempt per lapse until they are spuriously dead-lettered without ever failing. Postponing each blocked delivery back to the queue removes the held state entirely and inherits hold's retry exemption; the costs are a ~1s release quantization and the released message being a fresh attempt rather than the same one. - **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. diff --git a/doc/rfc/consumer-hold.md b/doc/rfc/consumer-hold.md index af7edc1c..fb167dc9 100644 --- a/doc/rfc/consumer-hold.md +++ b/doc/rfc/consumer-hold.md @@ -65,7 +65,7 @@ The postpone wording binds only what every plausible backend can express. The re - **A non-blocking postpone that lets later messages flow past the held one.** If the partition should keep flowing, success already says that — acknowledge and let the next message carry the work. The only reason to hold is that the partition must wait, so the primitive blocks; offering both semantics doubles the contract for a case with no user. - **Signaling hold through the processing result** — a sentinel error breaks the errors-are-failures rule, and a richer result type rewrites every controller and mock for one field. - **A framework-imposed maximum hold horizon.** Dead-letters lawful long waits; bounds belong to the domain that understands them. -- **Driving the wait through the consumer gate's write surface from inside controllers.** The gate parks deliveries in flight ahead of processing, so long waits inherit the parking costs above plus an unsolved recovery story when the opening signal is missed; it stays what it is — a stop/observe/start lever for tests and operators. The two are easy to tell apart: the gate is an external, event-ended stop — someone stopping the controller at the door before it sees the message; hold is a controller-chosen, timer-ended wait — the controller saw the work and decided to come back later. +- **Driving the wait through the consumer gate's write surface from inside controllers.** The gate is the same postpone mechanism applied *before* the controller by an *external* decision — a stop/observe/start lever for tests and operators, opened from outside. A controller's own wait needs no gate state to flip: it holds its delivery directly. The two are easy to tell apart: the gate is someone stopping the controller at the door before it sees the message; hold is the controller seeing the work and deciding to come back later. ## Follow-ups diff --git a/doc/rfc/index.md b/doc/rfc/index.md index 37e0d867..cfa743a6 100644 --- a/doc/rfc/index.md +++ b/doc/rfc/index.md @@ -6,7 +6,7 @@ Design documents and technical proposals, grouped by scope. Shared/cross-cutting - [SQL-Based Distributed Queue](sql-queue-rfc.md) - MySQL-based distributed message queue with partition leasing and at-least-once delivery (used by SubmitQueue, Stovepipe, and other repo-local services) - [Message Queue Contract](messagequeue-contract.md) - How queue payloads are defined (Protobuf, serialized as protobuf JSON), located by audience (external in `api/{domain}/messagequeue/`, internal in `{domain}/core/messagequeue/`), bound to topics (the `topics` proto option), and enforced by Bazel visibility -- [Consumer Gate](consumer-gate.md) - Stopping and starting individual queue controllers at runtime via consumer middleware: parked deliveries held in-flight with visibility extension, gate state as a separate extension with a file-based first implementation shared by tests and operators +- [Consumer Gate](consumer-gate.md) - Stopping and starting individual queue controllers at runtime via a consumer-side check: blocked deliveries are recorded as parked and postponed back to the queue (re-checked on redelivery), gate state as a separate extension with a file-based first implementation shared by tests and operators - [Consumer Hold](consumer-hold.md) - Fourth delivery outcome letting a controller postpone its delivery: the message becomes a partition barrier that pauses consumption for a chosen delay, redelivers in order, and does not count as a failure toward dead-lettering - [Change URIs](change-uri.md) - Identity of a code change: `scheme://{host[:port]}/{path}` per provider (GitHub PR, Phabricator Diff, git ref/commit) and canonical-form rules diff --git a/platform/consumer/README.md b/platform/consumer/README.md index ba6ea576..0c4678a4 100644 --- a/platform/consumer/README.md +++ b/platform/consumer/README.md @@ -132,9 +132,9 @@ Several mechanisms can delay work; they mean different things. Pick by what you' | "This delivery failed — retry it" | return a retryable error (framework nacks) | keeps flowing — a failure never halts its partition | | "I'm still working — keep my lease" | `delivery.ExtendVisibilityTimeout(...)` | blocked behind the in-flight delivery | | "Done for now — wake this partition in N ms" | `delivery.Hold(N)` then `return nil` | paused behind the postponed message (barrier), redelivers first in order | -| "Stop this controller/partition from outside" (tests, operators) | consumer gate (`platform/extension/consumergate`) | parked in flight until the gate opens | +| "Stop this controller/partition from outside" (tests, operators) | consumer gate (`platform/extension/consumergate`) | paused — blocked deliveries are parked + postponed until the gate opens | -Gate vs hold, since both pause a partition: the **gate** is an external, event-ended stop — someone stops the controller at the door, before `Process` ever sees the message. **Hold** is a controller-chosen, timer-ended wait — the controller saw the work and decided to come back later. Business logic never closes or opens gates; a controller that needs to back off uses hold. +Gate vs hold, since both pause a partition through the same postpone mechanism: the **gate** is an external stop — someone stops the controller at the door, before `Process` ever sees the message, and the wait ends when they open it. **Hold** is a controller-chosen wait — the controller saw the work and decided to come back later, and the wait ends on its own timer. Business logic never closes or opens gates; a controller that needs to back off uses hold. ## Lifecycle diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index 937363c2..80921215 100644 --- a/platform/consumer/consumer.go +++ b/platform/consumer/consumer.go @@ -34,15 +34,10 @@ const ( // 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 + // defaultGateRecheckDelayMs is how long a gate-blocked delivery is + // postponed before it redelivers and re-checks the gate (milliseconds). + // It bounds the release latency after a gate opens. + defaultGateRecheckDelayMs = int64(1000) ) // Consumer orchestrates multiple queue consumers. It handles subscription lifecycle, @@ -74,10 +69,11 @@ type consumer struct { 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 + // gateRecheckDelayMs is how long a gate-blocked delivery is postponed + // before it redelivers and re-checks the gate. Fixed to + // defaultGateRecheckDelayMs by New; a field (not the const) so in-package + // tests can exercise the re-check path quickly. + gateRecheckDelayMs int64 mu sync.Mutex stopped bool @@ -116,7 +112,7 @@ func New(logger *zap.SugaredLogger, scope tally.Scope, registry TopicRegistry, p registry: registry, processor: processor, gate: gate, - gateExtendInterval: defaultGateExtendInterval, + gateRecheckDelayMs: defaultGateRecheckDelayMs, subscriptions: make(map[TopicKey]*activeSubscription), } } @@ -367,11 +363,12 @@ 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) { + // Consumer gate: a delivery whose gate is closed is recorded as parked and + // postponed (barrier + re-check on redelivery); a false return also covers + // shutdown-while-checking, where the delivery is left in flight so its + // visibility lapses into a normal redelivery. Either way there is nothing + // further to do here. Gate read errors fail open inside checkGate. + if !m.checkGate(ctx, controller, delivery, controllerScope) { return } @@ -540,22 +537,18 @@ 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. +// checkGate clears a delivery through the consumer gate before it reaches the +// controller. It returns true when the delivery may proceed, false when the +// gate handled it: a blocked delivery is recorded as parked and postponed, so +// the same message redelivers after the re-check delay and re-enters here — +// the gate never waits in memory and never holds a lease. // -// 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 { +// Failures fail open: if gate state cannot be read, processing proceeds and +// the failure is surfaced via logs and metrics. Park is best-effort (the +// record is observability, never the outcome), and a failed postpone is +// abandoned like a failed ack — the visibility timeout lapses into a normal +// redelivery, which re-checks the gate. +func (m *consumer) checkGate(ctx context.Context, controller Controller, delivery extqueue.Delivery, scope tally.Scope) bool { const opName = "gate" msg := delivery.Message() @@ -565,6 +558,8 @@ func (m *consumer) waitGate(ctx context.Context, controller Controller, delivery entry, err := m.gate.Enter(ctx, consumergate.Key{ConsumerGroup: consumerGroup, PartitionKey: msg.PartitionKey}) if err != nil { if errors.Is(err, context.Canceled) { + // Shutting down: leave the delivery in flight; visibility lapses + // into a normal redelivery. return false } metrics.NamedCounter(scope, opName, "enter_errors", 1) @@ -576,14 +571,6 @@ func (m *consumer) waitGate(ctx context.Context, controller Controller, delivery ) 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, @@ -592,49 +579,48 @@ func (m *consumer) waitGate(ctx context.Context, controller Controller, delivery 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", + if !entry.Blocked() { + // Unconditional best-effort cleanup: if this delivery was parked on an + // earlier re-check, the gate has opened and the record must go so + // observers see an empty parked set. A no-op when never parked. + if unparkErr := entry.Unpark(ctx, descriptor); unparkErr != nil { + metrics.NamedCounter(scope, opName, "unpark_errors", 1) + m.logger.Warnw("failed to remove parked record on admit", "consumer_group", consumerGroup, "topic", topic, "message_id", msg.ID, - "error", waitErr, + "error", unparkErr, ) - 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 - } } + return true + } + + // Blocked: record the observation, then postpone the delivery so the + // partition waits behind it (barrier) and the gate is re-checked on + // redelivery without burning retry budget. + if parkErr := entry.Park(ctx, descriptor); parkErr != nil { + metrics.NamedCounter(scope, opName, "park_errors", 1) + m.logger.Warnw("failed to write parked record, postponing anyway", + "consumer_group", consumerGroup, + "topic", topic, + "message_id", msg.ID, + "error", parkErr, + ) + } + + metrics.NamedCounter(scope, opName, "parked", 1) + postponeOp := metrics.Begin(scope, "postpone", metrics.StorageLatencyBuckets) + postponeErr := delivery.Postpone(ctx, m.gateRecheckDelayMs) + postponeOp.Complete(postponeErr) + if postponeErr != nil { + m.logger.Errorw("failed to postpone gated delivery, leaving in flight", + "consumer_group", consumerGroup, + "topic", topic, + "message_id", msg.ID, + "error", postponeErr, + ) } + return false } func controllerClassificationTags(err error) []metrics.Tag { diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go index 2a432a19..0e6e5cc1 100644 --- a/platform/consumer/consumer_test.go +++ b/platform/consumer/consumer_test.go @@ -1122,49 +1122,34 @@ func TestConsumer_ConsumeLoopSurvivesCallerDeadline(t *testing.T) { } // fakeGate is a channel-instrumented consumergate.Gate so tests can await the -// park/release transitions instead of sleeping. +// park/unpark transitions instead of sleeping. type fakeGate struct { - mu sync.Mutex - closed map[consumergate.Key]bool - changed chan struct{} - err error + mu sync.Mutex + closed map[consumergate.Key]bool + err error parked chan consumergate.Parked - released chan string // message IDs + unparked 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), + unparked: 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) { @@ -1181,45 +1166,29 @@ func (f *fakeGate) isClosed(consumerGroup, partitionKey string) bool { } // 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. +// returns an entry whose Blocked reflects the gate state at Enter time. Park +// stamps the entered identity onto the descriptor and announces it on the +// parked channel; Unpark announces the message ID on the unparked 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 + return &fakeEntry{gate: f, key: key, blocked: f.isClosed(key.ConsumerGroup, key.PartitionKey)}, nil } -// fakeBlockedEntry is the entry handed out for a closed fake gate. -type fakeBlockedEntry struct { - gate *fakeGate - key consumergate.Key +// fakeEntry is the entry handed out by fakeGate.Enter. +type fakeEntry struct { + gate *fakeGate + key consumergate.Key + blocked bool } -func (*fakeBlockedEntry) Blocked() bool { return true } +func (e *fakeEntry) Blocked() bool { return e.blocked } -func (e *fakeBlockedEntry) Watch(ctx context.Context, descriptor consumergate.DeliveryDescriptor) <-chan error { - parked := consumergate.Parked{ +func (e *fakeEntry) Park(_ context.Context, descriptor consumergate.DeliveryDescriptor) error { + e.gate.parked <- consumergate.Parked{ ConsumerGroup: e.key.ConsumerGroup, Topic: descriptor.Topic, MessageID: descriptor.MessageID, @@ -1227,32 +1196,12 @@ func (e *fakeBlockedEntry) Watch(ctx context.Context, descriptor consumergate.De 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 - } + return nil +} - select { - case <-ctx.Done(): - ch <- ctx.Err() - return - case <-changed: - } - } - }() - return ch +func (e *fakeEntry) Unpark(_ context.Context, descriptor consumergate.DeliveryDescriptor) error { + e.gate.unparked <- descriptor.MessageID + return nil } // startGatedConsumer builds a consumer with the fake gate directly as the 5th @@ -1280,13 +1229,24 @@ func startGatedConsumer(t *testing.T, ctrl *gomock.Controller, gate consumergate 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{}) { +// gatedDelivery builds a MockDelivery for a delivery a closed gate will +// postpone: the returned channel closes when the framework postpones it with +// the gate re-check delay. No Ack/Nack/Reject expectations are set, so any of +// those calls fails the test. +func gatedDelivery(t *testing.T, ctrl *gomock.Controller, msg entityqueue.Message) (*queuemock.MockDelivery, chan struct{}) { + t.Helper() mockDel := queuemock.NewMockDelivery(ctrl) - done := setupDelivery(mockDel, msg, nil, nil) - mockDel.EXPECT().ExtendVisibilityTimeout(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() - return mockDel, done + mockDel.EXPECT().Message().Return(msg).AnyTimes() + mockDel.EXPECT().Attempt().Return(1).AnyTimes() + mockDel.EXPECT().ReceivedAt().Return(time.Now().UnixMilli()).AnyTimes() + mockDel.EXPECT().Metadata().Return(nil).AnyTimes() + mockDel.EXPECT().DeliveryID().Return(msg.ID).AnyTimes() + postponed := make(chan struct{}) + mockDel.EXPECT().Postpone(gomock.Any(), defaultGateRecheckDelayMs).DoAndReturn(func(context.Context, int64) error { + close(postponed) + return nil + }) + return mockDel, postponed } func TestConsumer_Gate_OpenGatePassesThrough(t *testing.T) { @@ -1300,7 +1260,8 @@ func TestConsumer_Gate_OpenGatePassesThrough(t *testing.T) { }) msg := entityqueue.NewMessage("msg-1", []byte("payload"), "partition1", nil) - mockDel, done := gatedDelivery(ctrl, msg) + mockDel := queuemock.NewMockDelivery(ctrl) + done := setupDelivery(mockDel, msg, nil, nil) deliveryChan <- mockDel <-done @@ -1311,7 +1272,7 @@ func TestConsumer_Gate_OpenGatePassesThrough(t *testing.T) { require.NoError(t, c.Stop(30000)) } -func TestConsumer_Gate_ParksThenReleases(t *testing.T) { +func TestConsumer_Gate_BlockedParksAndPostpones(t *testing.T) { ctrl := gomock.NewController(t) gate := newFakeGate() gate.close(consumergate.Key{ConsumerGroup: "test-group"}) @@ -1323,11 +1284,12 @@ func TestConsumer_Gate_ParksThenReleases(t *testing.T) { }) msg := entityqueue.NewMessage("msg-1", []byte("payload"), "partition1", nil) - mockDel, done := gatedDelivery(ctrl, msg) + mockDel, postponed := gatedDelivery(t, 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. + // The parked record is written before the delivery is postponed, 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) @@ -1335,14 +1297,21 @@ func TestConsumer_Gate_ParksThenReleases(t *testing.T) { assert.Equal(t, "partition1", parked.PartitionKey) assert.Equal(t, []byte("payload"), parked.Payload) assert.Equal(t, 1, parked.Attempt) + + // The delivery is postponed with the re-check delay and never processed. + <-postponed 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. + // Open the gate and feed the redelivery (postpone finalizes the original + // delivery; the queue redelivers the same message as a fresh attempt). + // The redelivery unparks the record and is processed and acked. gate.open(consumergate.Key{ConsumerGroup: "test-group"}) - assert.Equal(t, "msg-1", <-gate.released) + redelivery := queuemock.NewMockDelivery(ctrl) + done := setupDelivery(redelivery, msg, nil, nil) + deliveryChan <- redelivery <-done assert.True(t, processed.Load()) + assert.Equal(t, "msg-1", <-gate.unparked, "the admit path must remove the parked record") require.NoError(t, c.Stop(30000)) } @@ -1359,16 +1328,18 @@ func TestConsumer_Gate_PartitionScoped(t *testing.T) { }) gatedMsg := entityqueue.NewMessage("gated-msg", []byte("p"), "gated-partition", nil) - gatedDel, gatedDone := gatedDelivery(ctrl, gatedMsg) + gatedDel, gatedPostponed := gatedDelivery(t, ctrl, gatedMsg) openMsg := entityqueue.NewMessage("open-msg", []byte("p"), "open-partition", nil) - openDel, openDone := gatedDelivery(ctrl, openMsg) + openDel := queuemock.NewMockDelivery(ctrl) + openDone := setupDelivery(openDel, openMsg, nil, nil) deliveryChan <- gatedDel parked := <-gate.parked assert.Equal(t, "gated-msg", parked.MessageID) + <-gatedPostponed // Unrelated traffic keeps flowing through the same controller while one - // partition is parked. + // partition is gated. deliveryChan <- openDel <-openDone _, ok := handled.Load("open-msg") @@ -1376,7 +1347,11 @@ func TestConsumer_Gate_PartitionScoped(t *testing.T) { _, ok = handled.Load("gated-msg") assert.False(t, ok) + // Open the gate; the redelivery of the gated message processes. gate.open(consumergate.Key{ConsumerGroup: "test-group", PartitionKey: "gated-partition"}) + redelivery := queuemock.NewMockDelivery(ctrl) + gatedDone := setupDelivery(redelivery, gatedMsg, nil, nil) + deliveryChan <- redelivery <-gatedDone _, ok = handled.Load("gated-msg") assert.True(t, ok) @@ -1384,7 +1359,7 @@ func TestConsumer_Gate_PartitionScoped(t *testing.T) { require.NoError(t, c.Stop(30000)) } -func TestConsumer_Gate_ShutdownWhileParked(t *testing.T) { +func TestConsumer_Gate_StopWhileGated(t *testing.T) { ctrl := gomock.NewController(t) gate := newFakeGate() gate.close(consumergate.Key{ConsumerGroup: "test-group"}) @@ -1396,16 +1371,15 @@ func TestConsumer_Gate_ShutdownWhileParked(t *testing.T) { }) msg := entityqueue.NewMessage("msg-1", []byte("payload"), "partition1", nil) - mockDel, _ := gatedDelivery(ctrl, msg) + mockDel, postponed := gatedDelivery(t, ctrl, msg) deliveryChan <- mockDel - <-gate.parked + <-postponed - // 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. + // Nothing is held in memory while a gate is closed — the delivery was + // postponed back to the queue — so Stop must not stall and the controller + // must never have run. 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) { @@ -1421,7 +1395,8 @@ func TestConsumer_Gate_FailsOpenOnReadError(t *testing.T) { }) msg := entityqueue.NewMessage("msg-1", []byte("payload"), "partition1", nil) - mockDel, done := gatedDelivery(ctrl, msg) + mockDel := queuemock.NewMockDelivery(ctrl) + done := setupDelivery(mockDel, msg, nil, nil) deliveryChan <- mockDel <-done diff --git a/platform/extension/consumergate/README.md b/platform/extension/consumergate/README.md index 244ed9d9..4aadb95c 100644 --- a/platform/extension/consumergate/README.md +++ b/platform/extension/consumergate/README.md @@ -4,20 +4,21 @@ Runtime stop/start of individual queue controllers without stopping the service ## 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. +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 the admission check and the parked-delivery observation records; it never waits. `Enter` checks a delivery's gate key synchronously and returns an `Entry`. When the entry is blocked, the caller records the observation with `Park` (the implementation stamps the entered identity and `ParkedAtMs`) and defers the delivery itself — the consumer postpones it, so the same message redelivers after a re-check delay and passes through `Enter` again. When the gate has opened, the caller removes the record with `Unpark` (a no-op when nothing was parked, so the admit path may call it unconditionally) and proceeds. 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 gate does not own the source queue delivery and cannot acknowledge, nack, reject, postpone, remove, or move it. Deferring a blocked delivery is the caller's action; the gate only records what is waiting. -The package defines three interfaces, the package-level `Wait` helper, plus the `Config`: +The package defines three interfaces: -- `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. +- `Gate` exposes `Enter`, a synchronous check keyed on consumer group and partition that returns an `Entry`. +- `Entry` exposes `Blocked`, plus `Park`/`Unpark` over a `DeliveryDescriptor` containing only caller-owned message data; the implementation combines that descriptor with the gate identity captured by `Enter` to create (or remove) the observable `Parked` record. Re-parking the same delivery on a re-check overwrites its record, so records stay bounded by currently blocked deliveries. 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. +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). The record is removed on the admit path once the gate opens, 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. +An `Enter`, `Park`, or `Unpark` that cannot read or write 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 diff --git a/platform/extension/consumergate/consumergate.go b/platform/extension/consumergate/consumergate.go index 0a646c44..581bf755 100644 --- a/platform/extension/consumergate/consumergate.go +++ b/platform/extension/consumergate/consumergate.go @@ -18,23 +18,18 @@ // 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. +// delivery immediately, while a closed gate stops it. The gate never waits — +// when an Entry is blocked, the caller records the observable Parked record +// with Entry.Park and postpones the delivery, so the same message redelivers +// after a re-check delay and passes through Enter again; when the gate has +// opened, the caller removes the record with Entry.Unpark and proceeds. The +// gate owns the parked-delivery observation records (stamping ParkedAtMs and +// the gate identity captured by Enter); the caller owns the delivery outcome. // // 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. +// interfaces) and Admin (the write surface used by tests and tooling). +// 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 @@ -104,7 +99,8 @@ type Parked struct { Attempt int // ParkedAtMs is when the delivery was parked (Unix milliseconds). Stamped - // by the gate implementation when it actually blocks. + // by the gate implementation on Park; re-parking the same delivery + // refreshes it. ParkedAtMs int64 } @@ -113,10 +109,9 @@ type Parked struct { 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. + // returned Entry is unblocked and the delivery may proceed at once. When + // the gate is closed, the returned Entry is blocked; the caller parks the + // delivery's record and defers the delivery itself. // // 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. @@ -126,41 +121,21 @@ type Gate interface { // 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. + // An unblocked entry needs no Park — 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) + // Park records the delivery as parked, adding the gate identity captured + // by Enter and the implementation-owned ParkedAtMs. Parking the same + // delivery again (on a re-check of a still-closed gate) overwrites the + // previous record, so records stay bounded by currently blocked + // deliveries. Park observes only the descriptor — it does not own or + // mutate the source queue delivery. + Park(ctx context.Context, descriptor DeliveryDescriptor) error + + // Unpark removes the delivery's parked record, if one exists. Unparking a + // delivery that was never parked is a no-op, so callers may invoke it + // unconditionally on the admit path. + Unpark(ctx context.Context, descriptor DeliveryDescriptor) error } // Admin is the write surface used by tests and tooling to operate gates and @@ -177,25 +152,3 @@ type Admin interface { // 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/README.md b/platform/extension/consumergate/file/README.md index 14646973..78147c6d 100644 --- a/platform/extension/consumergate/file/README.md +++ b/platform/extension/consumergate/file/README.md @@ -10,9 +10,9 @@ Stores gate state as plain files under an explicitly configured root directory f {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. +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 refreshed on each re-check of a still-closed gate and removed on the admit path once the gate opens, so the parked tree contains only currently blocked deliveries and does not retain payloads after release. 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. +Gate state is not cached and the store never polls. Each delivery attempt reads its applicable gate files; a blocked delivery's re-check cadence is the consumer's postpone delay, so gate state is re-read on every redelivery. ## Operating it by hand diff --git a/platform/extension/consumergate/file/store.go b/platform/extension/consumergate/file/store.go index ac060945..cbe8c596 100644 --- a/platform/extension/consumergate/file/store.go +++ b/platform/extension/consumergate/file/store.go @@ -26,20 +26,12 @@ // 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. +// Enter reads the applicable gate files for every delivery; the store itself +// never waits or polls. A blocked delivery's re-check cadence is owned by the +// caller (the consumer postpones the delivery and re-enters on redelivery), so +// gate state is re-read on every attempt regardless of filesystem event +// support — which varies across bind mounts, overlay or network filesystems, +// rootless Docker, and Docker Desktop's host/container filesystem bridge. // // 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 @@ -64,8 +56,7 @@ import ( // Store implements consumergate.Gate and consumergate.Admin over a directory. type Store struct { - dir string - pollInterval time.Duration + dir string } // Verify interface compliance at compile time. @@ -76,17 +67,9 @@ var ( // 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, - } +// parked records", and writes create what they need. +func New(dir string) *Store { + return &Store{dir: dir} } // gatePath returns the gate file path for a key: the "all" marker when the key @@ -126,59 +109,34 @@ func (s *Store) isGated(consumerGroup, partitionKey string) (bool, error) { } // 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. +// gate identified by key is open, and a blocked Entry 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 + return entry{store: s, key: key, blocked: gated}, nil } -// parkedEntry is the Entry for a delivery held by a closed gate. -type parkedEntry struct { - // store is the file store that gated the delivery. +// entry is the Entry for one delivery's gate check. +type entry struct { + // store is the file store that performed the check. store *Store // key is the gate identity the delivery entered with. key consumergate.Key + // blocked records whether the gate was closed at Enter time. + blocked bool } // Blocked implements consumergate.Entry. -func (*parkedEntry) Blocked() bool { return true } +func (e entry) Blocked() bool { return e.blocked } -// 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{ +// Park implements consumergate.Entry. It writes the parked record, combining +// the caller's delivery description with the gate identity captured by Enter +// and a fresh ParkedAtMs stamp. Re-parking the same delivery overwrites the +// previous record. +func (e entry) Park(_ context.Context, descriptor consumergate.DeliveryDescriptor) error { + return e.store.recordParked(consumergate.Parked{ ConsumerGroup: e.key.ConsumerGroup, Topic: descriptor.Topic, MessageID: descriptor.MessageID, @@ -186,41 +144,13 @@ func (e *parkedEntry) Watch(ctx context.Context, descriptor consumergate.Deliver 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 +// Unpark implements consumergate.Entry. Removing an absent record is a no-op, +// so callers may invoke it unconditionally on the admit path. +func (e entry) Unpark(_ context.Context, descriptor consumergate.DeliveryDescriptor) error { + return e.store.removeParked(e.key.ConsumerGroup, descriptor.Topic, descriptor.MessageID) } // recordParked writes a parked-delivery record. Re-recording the same delivery diff --git a/platform/extension/consumergate/file/store_test.go b/platform/extension/consumergate/file/store_test.go index 32ff8288..1ab0abf0 100644 --- a/platform/extension/consumergate/file/store_test.go +++ b/platform/extension/consumergate/file/store_test.go @@ -19,32 +19,12 @@ import ( "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() @@ -107,7 +87,7 @@ func TestIsGated(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - store := New(t.TempDir(), testCfg) + store := New(t.TempDir()) for _, key := range tt.close { require.NoError(t, store.Close(ctx, key, consumergate.Metadata{Reason: "test", CreatedBy: "unit", CreatedAtMs: 1})) } @@ -120,7 +100,7 @@ func TestIsGated(t *testing.T) { func TestOpenClosesGate(t *testing.T) { ctx := context.Background() - store := New(t.TempDir(), testCfg) + store := New(t.TempDir()) key := consumergate.Key{ConsumerGroup: "orchestrator-batch", PartitionKey: "queue-a"} require.NoError(t, store.Close(ctx, key, consumergate.Metadata{Reason: "pause", CreatedBy: "unit", CreatedAtMs: 1})) @@ -138,14 +118,14 @@ func TestOpenClosesGate(t *testing.T) { } func TestCloseRequiresConsumerGroup(t *testing.T) { - store := New(t.TempDir(), testCfg) + store := New(t.TempDir()) 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) + store := New(t.TempDir()) parked := consumergate.Parked{ ConsumerGroup: "runway-mergeconflictcheck", @@ -181,7 +161,7 @@ func TestParkedRecordLifecycle(t *testing.T) { } func TestListParkedEmpty(t *testing.T) { - store := New(t.TempDir(), testCfg) + store := New(t.TempDir()) records, err := store.ListParked(context.Background(), "no-such-group") require.NoError(t, err) assert.Empty(t, records) @@ -190,7 +170,7 @@ func TestListParkedEmpty(t *testing.T) { func TestListParkedSkipsTempFiles(t *testing.T) { ctx := context.Background() dir := t.TempDir() - store := New(dir, testCfg) + store := New(dir) parked := consumergate.Parked{ ConsumerGroup: "group", @@ -211,7 +191,7 @@ func TestListParkedSkipsTempFiles(t *testing.T) { } func TestMissingDirIsNotGated(t *testing.T) { - store := New(filepath.Join(t.TempDir(), "does-not-exist"), testCfg) + store := New(filepath.Join(t.TempDir(), "does-not-exist")) gated, err := store.isGated("group", "part") require.NoError(t, err) assert.False(t, gated) @@ -219,48 +199,51 @@ func TestMissingDirIsNotGated(t *testing.T) { func TestEnter_OpenGateUnblocked(t *testing.T) { ctx := context.Background() - store := New(t.TempDir(), testCfg) + store := New(t.TempDir()) - 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{ + descriptor := consumergate.DeliveryDescriptor{ Topic: "topic", MessageID: "msg-1", Payload: []byte("hello"), Attempt: 1, - })) + } + + entry, err := store.Enter(ctx, consumergate.Key{ConsumerGroup: "group", PartitionKey: "part"}) + require.NoError(t, err) + assert.False(t, entry.Blocked()) + + // Unparking a never-parked delivery is a no-op on the admit path. + require.NoError(t, entry.Unpark(ctx, descriptor)) - // 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) { +func TestEnter_ClosedGateParkAndRelease(t *testing.T) { ctx := context.Background() - store := New(t.TempDir(), testCfg) + store := New(t.TempDir()) key := consumergate.Key{ConsumerGroup: "group"} require.NoError(t, store.Close(ctx, key, consumergate.Metadata{Reason: "test", CreatedBy: "unit", CreatedAtMs: 1})) + descriptor := consumergate.DeliveryDescriptor{ + Topic: "topic", + MessageID: "msg-1", + Payload: []byte("hello"), + Attempt: 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, - }) - }() + // Park records the delivery; the caller supplies only the delivery + // content, the store stamps the entered identity and ParkedAtMs. + require.NoError(t, entry.Park(ctx, descriptor)) - records := awaitParked(t, store, ctx, "group") + records, err := store.ListParked(ctx, "group") + require.NoError(t, err) require.Len(t, records, 1) assert.Equal(t, "group", records[0].ConsumerGroup) assert.Equal(t, "part", records[0].PartitionKey) @@ -270,52 +253,21 @@ func TestEnter_ClosedGateParksThenReleases(t *testing.T) { 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) - + // Re-parking on a re-check overwrites the record, not duplicates it. + descriptor.Attempt = 1 // postponed redeliveries restart at attempt 1 + require.NoError(t, entry.Park(ctx, descriptor)) 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})) + require.Len(t, records, 1) - entry, err := store.Enter(ctx, consumergate.Key{ConsumerGroup: "group", PartitionKey: "part"}) + // Open the gate; the next Enter is unblocked and Unpark removes the record. + require.NoError(t, store.Open(ctx, key)) + entry, err = store.Enter(ctx, consumergate.Key{ConsumerGroup: "group", PartitionKey: "part"}) require.NoError(t, err) - require.True(t, entry.Blocked()) + require.False(t, entry.Blocked()) + require.NoError(t, entry.Unpark(ctx, descriptor)) - 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") + records, err = store.ListParked(ctx, "group") require.NoError(t, err) assert.Empty(t, records) } @@ -325,7 +277,7 @@ func TestEnter_MediumError(t *testing.T) { dir := filepath.Join(t.TempDir(), "not-a-dir") require.NoError(t, os.WriteFile(dir, []byte("x"), 0o644)) - store := New(dir, testCfg) + store := New(dir) _, err := store.Enter(context.Background(), consumergate.Key{ConsumerGroup: "group", PartitionKey: "part"}) require.Error(t, err) } diff --git a/platform/extension/consumergate/mock/consumergate_mock.go b/platform/extension/consumergate/mock/consumergate_mock.go index c956d9d0..3c5e01f5 100644 --- a/platform/extension/consumergate/mock/consumergate_mock.go +++ b/platform/extension/consumergate/mock/consumergate_mock.go @@ -94,18 +94,32 @@ func (mr *MockEntryMockRecorder) Blocked() *gomock.Call { 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 { +// Park mocks base method. +func (m *MockEntry) Park(ctx context.Context, descriptor consumergate.DeliveryDescriptor) error { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Watch", ctx, descriptor) - ret0, _ := ret[0].(<-chan error) + ret := m.ctrl.Call(m, "Park", ctx, descriptor) + ret0, _ := ret[0].(error) return ret0 } -// Watch indicates an expected call of Watch. -func (mr *MockEntryMockRecorder) Watch(ctx, descriptor any) *gomock.Call { +// Park indicates an expected call of Park. +func (mr *MockEntryMockRecorder) Park(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) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Park", reflect.TypeOf((*MockEntry)(nil).Park), ctx, descriptor) +} + +// Unpark mocks base method. +func (m *MockEntry) Unpark(ctx context.Context, descriptor consumergate.DeliveryDescriptor) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Unpark", ctx, descriptor) + ret0, _ := ret[0].(error) + return ret0 +} + +// Unpark indicates an expected call of Unpark. +func (mr *MockEntryMockRecorder) Unpark(ctx, descriptor any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Unpark", reflect.TypeOf((*MockEntry)(nil).Unpark), ctx, descriptor) } // MockAdmin is a mock of Admin interface. @@ -174,42 +188,3 @@ 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/gate.go b/platform/extension/consumergate/noop/gate.go index fa4c37a0..7d646fb2 100644 --- a/platform/extension/consumergate/noop/gate.go +++ b/platform/extension/consumergate/noop/gate.go @@ -45,11 +45,9 @@ func (g Gate) Enter(_ context.Context, _ consumergate.Key) (consumergate.Entry, // 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 -} +// Park implements consumergate.Entry. A no-op gate records nothing. It is +// never reached in practice because Blocked reports false. +func (Gate) Park(context.Context, consumergate.DeliveryDescriptor) error { return nil } + +// Unpark implements consumergate.Entry. A no-op gate holds no records. +func (Gate) Unpark(context.Context, consumergate.DeliveryDescriptor) error { return nil } diff --git a/platform/extension/consumergate/noop/gate_test.go b/platform/extension/consumergate/noop/gate_test.go index ae5ea67c..6a8c7df0 100644 --- a/platform/extension/consumergate/noop/gate_test.go +++ b/platform/extension/consumergate/noop/gate_test.go @@ -28,10 +28,13 @@ func TestGate_EnterNeverBlocks(t *testing.T) { 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{ + + descriptor := consumergate.DeliveryDescriptor{ Topic: "topic", MessageID: "msg-1", Payload: []byte("hello"), Attempt: 1, - })) + } + require.NoError(t, entry.Park(context.Background(), descriptor)) + require.NoError(t, entry.Unpark(context.Background(), descriptor)) } diff --git a/service/runway/server/main.go b/service/runway/server/main.go index cb121315..beba8ca0 100644 --- a/service/runway/server/main.go +++ b/service/runway/server/main.go @@ -372,5 +372,5 @@ func newConsumerGate(logger *zap.Logger) consumergate.Gate { return consumergatenoop.New() } logger.Info("consumer gate configured", zap.String("dir", dir)) - return consumergatefile.New(dir, consumergate.DefaultConfig()) + return consumergatefile.New(dir) } diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index 6cd20701..8186b707 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -455,5 +455,5 @@ func newConsumerGate(logger *zap.Logger) consumergate.Gate { return consumergatenoop.New() } logger.Info("consumer gate configured", zap.String("dir", dir)) - return consumergatefile.New(dir, consumergate.DefaultConfig()) + return consumergatefile.New(dir) } diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 0c5bbc61..1fb01444 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -312,7 +312,7 @@ func newConsumerGate(logger *zap.Logger) consumergate.Gate { return consumergatenoop.New() } logger.Info("consumer gate configured", zap.String("dir", dir)) - return consumergatefile.New(dir, consumergate.DefaultConfig()) + return consumergatefile.New(dir) } // newChangeProvider creates a routing ChangeProvider containing GitHub and Phab ChangeProviders. diff --git a/test/e2e/submitqueue/harness_test.go b/test/e2e/submitqueue/harness_test.go index ad54fdc2..d8fb81c2 100644 --- a/test/e2e/submitqueue/harness_test.go +++ b/test/e2e/submitqueue/harness_test.go @@ -176,8 +176,9 @@ func (s *E2EIntegrationSuite) openGate(consumerGroup, partitionKey string) { // 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 +// The record is written by the gated service before it postpones the delivery +// (and refreshed on every re-check while the gate stays closed), so observing +// it proves the stopped controller caught 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() @@ -197,8 +198,8 @@ func (s *E2EIntegrationSuite) awaitParked(consumerGroup, topic, messageID string } // 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. +// The gated service removes the record when the redelivered message clears the +// open gate, so disappearance proves the delivery was admitted after the open. func (s *E2EIntegrationSuite) awaitUnparked(consumerGroup, topic, messageID string) { t := s.T() pollUntil(persistPollInterval, func() bool { diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go index 478bdc68..9dad3072 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/test/e2e/submitqueue/suite_test.go @@ -41,7 +41,6 @@ import ( 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" @@ -95,7 +94,7 @@ func (s *E2EIntegrationSuite) SetupSuite() { // 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()) + s.gate = consumergatefile.New(gateDir) // Use docker-compose from service/submitqueue (full stack), resolved from // the test runfiles. All three service images are built from a staged @@ -356,9 +355,9 @@ func (s *E2EIntegrationSuite) TestCancelRequest_InvalidSqid() { // 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. +// 5. Start: open the gate. The postponed check redelivers within a re-check +// tick and clears the open gate, 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