feat(consumergate): runtime stop/start of queue controllers + deterministic e2e cancel test#375
feat(consumergate): runtime stop/start of queue controllers + deterministic e2e cancel test#375sbalabanov wants to merge 1 commit into
Conversation
|
|
| // | ||
| // opts configure optional behavior, e.g. WithGate to install the consumer-gate | ||
| // middleware. | ||
| func New(logger *zap.SugaredLogger, scope tally.Scope, registry TopicRegistry, processor errs.ErrorProcessor, opts ...Option) Consumer { |
There was a problem hiding this comment.
Pass Gate interface directly and not through options. Like other interface-typed arguments, it should never be null by convention and not need to check that. If no gates are required, the caller would wire up a no-op gate implementation. Please create one.
| func gateOptions(logger *zap.Logger) []consumer.Option { | ||
| dir := os.Getenv("CONSUMER_GATE_DIR") | ||
| if dir == "" { | ||
| return nil |
There was a problem hiding this comment.
prefer a default folder instead
| tmpName := tmp.Name() | ||
| if _, err := tmp.Write(data); err != nil { | ||
| tmp.Close() | ||
| os.Remove(tmpName) |
There was a problem hiding this comment.
do not swallow potential error, instead concatenate with errors.Join
also, redesign a code to not require to call os.Remove(tmpName) for every unsuccessful operation. Converge error flows and do it once.
| // controller, and false when the consumer is shutting down while parked — in | ||
| // that case the caller must not process, ack, or nack: extension stops, | ||
| // visibility lapses, and the queue redelivers normally. | ||
| func (g *gateMiddleware) hold(ctx context.Context, controller Controller, delivery extqueue.Delivery) bool { |
There was a problem hiding this comment.
Middleware always relies on periodic checks, however the potential implementation may be able to open the gate as soon as admin signal arrives if it supports notifications. Leave blocking the gate and unblocking to the consumer gate instead - you have to redesign the consumer gate interface to make it a part of the contract. The middleware in this case could be very thin or not needed itself, as Consumer can just check Gates open/close as part of its logic.
Please debate internally whether to implement recording part in the consumer or make it a part of the consumer gate interface responsibility. Make a judgement and implement the best variant.
…nistic e2e cancel test Implement doc/rfc/consumer-gate.md: - platform/extension/consumergate: extension contract — Gate, a single blocking Wait the consumer consults per delivery (implementations own the wait mechanism, so a notification-capable store can release instantly, and the parked-delivery observation records), Admin (write surface for tests and tooling), Config, Factory interface, and gomock mocks. - platform/extension/consumergate/file: polling implementation — gate state as plain files in a shared directory (gate file present = closed, rm = open), parked-delivery records as JSON stamped parked/released by the store, per-partition cached verdicts (TTL = poll interval) so an open gate costs no stat per message, and temp-file-plus-rename writes with converged error handling. - platform/extension/consumergate/noop: no-op gate for services and tests that do not need runtime gating. - platform/consumer: consumer.New takes the Gate as a required argument and consults it before each delivery. While a delivery is blocked the consumer keeps it in-flight by periodically extending visibility (no retry budget burned, partition order preserved). Shutdown while blocked leaves the delivery for normal redelivery; gate errors fail open with a log and counter. - Wiring: gateway, orchestrator (primary + DLQ), and runway construct the file-backed gate rooted at CONSUMER_GATE_DIR, defaulting to /var/submitqueue/consumergate — the path the compose stack bind-mounts into every service; stovepipe wires the no-op gate. - test/e2e/submitqueue: replace TestCancel_RecordsIntent with TestCancel_CaughtPreBatch_NeverLands — the stop→observe→start scenario from the RFC. The gate parks runway's merge-conflict check for the test queue's partition before landing, the parked record proves the request is held pre-batch, the cancel drives it terminal cancelled, and after the gate opens a sentinel request landing on the same partitions proves the stale check signal was consumed and dropped — the cancelled change is never batched and never lands. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
sbalabanov
force-pushed
the
consumer-gate-impl
branch
from
d04bf53 to
fb4fde8
Compare
3 participants
Implements the consumer-gate RFC (#374). Stacked on the RFC branch; review this PR for the implementation only.
What
platform/extension/consumergate— the extension contract:Gate(read side consulted by the consumer middleware: is this group/partition gated, record a parked delivery, stamp its release),Admin(write surface for tests and tooling: close, open, list parked),Config, theFactoryinterface, and gomock mocks.platform/extension/consumergate/file— the first implementation: gate state as plain files in a shared directory (gates/{group}/all,gates/{group}/p-{urlenc(partition)}; parked records atparked/{group}/{topic}/{urlenc(id)}.json). Presence of a gate file means closed,rmopens it; all writes are temp-file-plus-rename.platform/consumer— aWithGateoption installs gate middleware in front of every registered controller. Closed gate ⇒ the delivery is parked in place (record written first, visibility extended each refresh tick, retry count untouched, partition order preserved) until the gate opens or the consumer shuts down; shutdown while parked leaves the delivery in-flight for normal redelivery. Gate state is a cached poll per (group, partition) at ~1s; read failures fail open with a log + counter.CONSUMER_GATE_DIR; unset means the middleware is absent. The compose stack bind-mounts one shared host directory (SQ_CONSUMER_GATE_DIR, defaulting to/tmp/sq-consumergate) into every service.E2e cancellation test rewritten on the gate
TestCancel_RecordsIntent(which could only assert the synchronous "cancelling" intent, with a comment pointing at exactly this missing lever) is replaced byTestCancel_CaughtPreBatch_NeverLands, the RFC's stop → observe → start walk-through:runway-mergeconflictcheck, scoped to the test queue's partition, before landing — exact by construction.cancelled(request_log shows accepted → cancelling → cancelled; operating store showsRequestStateCancelled).landedproves the stale signal was consumed — at which point the cancelled request is asserted still terminal cancelled, never batched, never landed.The e2e target gains a
no-sandboxtag: the bind-mounted gate dir must be a host path the Docker daemon can see, and Bazel 8's hermetic sandbox/tmpis not.Testing
make test— 85/85 unit tests pass, including new middleware tests (park/release, partition scoping, shutdown-while-parked, fail-open, cache expiry) and file-store tests.make e2e-test— both suites pass; the service logs show the full sequence:delivery parked by consumer gate→request cancelled (not batched)→parked delivery released by consumer gate→skipping mergeconflict signal for halted request.make lint,make check-tidy,make check-gazelle,make check-mocksall green.🤖 Generated with Claude Code