feat(consumergate): runtime stop/start of queue controllers + deterministic e2e cancel test#375
Draft
sbalabanov wants to merge 1 commit into
Draft
feat(consumergate): runtime stop/start of queue controllers + deterministic e2e cancel test#375sbalabanov wants to merge 1 commit into
sbalabanov wants to merge 1 commit into
Conversation
|
|
sbalabanov
commented
Jul 16, 2026
sbalabanov
force-pushed
the
consumer-gate-impl
branch
from
July 16, 2026 00:38
d04bf53 to
fb4fde8
Compare
sbalabanov
commented
Jul 16, 2026
sbalabanov
force-pushed
the
rfc-consumer-gate
branch
from
July 16, 2026 20:43
7ff27fb to
1c25eb9
Compare
sbalabanov
force-pushed
the
consumer-gate-impl
branch
from
July 16, 2026 20:43
fb4fde8 to
0b06f08
Compare
behinddwalls
force-pushed
the
consumer-gate-impl
branch
from
July 16, 2026 22:00
0b06f08 to
6b0aa49
Compare
sbalabanov
commented
Jul 16, 2026
sbalabanov
force-pushed
the
consumer-gate-impl
branch
2 times, most recently
from
July 16, 2026 22:55
4e229a3 to
2db191e
Compare
sbalabanov
commented
Jul 16, 2026
…nistic e2e cancel test Implement doc/rfc/consumer-gate.md: - platform/extension/consumergate: extension contract — Gate.Enter checks a delivery against its gate synchronously and returns an Entry future: an open gate admits the delivery immediately, a closed gate hands out a blocked Entry whose Wait records the parked delivery and blocks until the gate opens (implementations own the wait mechanism, so a notification-capable store can release instantly, and the parked-delivery observation records). Admin is the write surface for tests and tooling; Config, Factory interface, and gomock mocks round out the contract. - platform/extension/consumergate/file: polling implementation — gate state as plain files in a shared directory (gate file present = closed, rm = open), parked-delivery records as JSON stamped parked/released by the store, per-partition cached verdicts (TTL = poll interval) so an open gate costs no stat per delivery, and temp-file-plus-rename writes with converged error handling. - platform/extension/consumergate/noop: no-op gate for services and tests that do not need runtime gating. - platform/consumer: consumer.New takes the Gate as a required argument and clears every delivery through Enter. Only a blocked delivery pays for queue mechanics: a lazily armed timer keeps it in-flight by periodically extending visibility (no goroutine unless actually blocked, no retry budget burned, partition order preserved), and blocked wait time is recorded on a latency histogram. Shutdown while blocked leaves the delivery for normal redelivery; gate errors and failed visibility extensions cancel the wait and fail open with a log and counter. - Wiring: gateway, orchestrator (primary + DLQ), and runway construct the file-backed gate rooted at CONSUMER_GATE_DIR, defaulting to /var/submitqueue/consumergate — the path the compose stack bind-mounts into every service; stovepipe wires the no-op gate. - test/e2e/submitqueue: replace TestCancel_RecordsIntent with TestCancel_CaughtPreBatch_NeverLands — the stop→observe→start scenario from the RFC. The gate parks runway's merge-conflict check for the test queue's partition before landing, the parked record proves the request is held pre-batch, the cancel drives it terminal cancelled, and after the gate opens a sentinel request landing on the same partitions proves the stale check signal was consumed and dropped — the cancelled change is never batched and never lands. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
sbalabanov
force-pushed
the
consumer-gate-impl
branch
from
July 16, 2026 23:36
2db191e to
617595b
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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