Skip to content

feat(consumergate): runtime stop/start of queue controllers + deterministic e2e cancel test #375

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
sbalabanov wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

feat(consumergate): runtime stop/start of queue controllers + deterministic e2e cancel test #375

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -364,7 +364,7 @@ local-stovepipe-stop: ## Stop the Stovepipe service

mocks: ## Generate mock files using mockgen
@echo "Generating mocks..."
@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/pusher/... ./submitqueue/extension/scorer/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/enumerator/... ./submitqueue/extension/speculation/dependencylimit/... ./submitqueue/extension/speculation/selector/... ./submitqueue/extension/speculation/selectionlimit/... ./submitqueue/extension/speculation/prioritizer/... ./submitqueue/extension/speculation/prioritizationlimit/... ./submitqueue/extension/validator/... ./submitqueue/extension/speculation/pathscorer/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/pusher/... ./submitqueue/extension/scorer/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/enumerator/... ./submitqueue/extension/speculation/dependencylimit/... ./submitqueue/extension/speculation/selector/... ./submitqueue/extension/speculation/selectionlimit/... ./submitqueue/extension/speculation/prioritizer/... ./submitqueue/extension/speculation/prioritizationlimit/... ./submitqueue/extension/validator/... ./submitqueue/extension/speculation/pathscorer/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
@echo "Mocks generated successfully!"

proto: ## Generate protobuf files from .proto definitions
Expand Down
6 changes: 5 additions & 1 deletion platform/consumer/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ go_library(
deps = [
"//platform/base/messagequeue:go_default_library",
"//platform/errs:go_default_library",
"//platform/extension/consumergate:go_default_library",
"//platform/extension/messagequeue:go_default_library",
"//platform/metrics:go_default_library",
"@com_github_uber_go_tally//:go_default_library",
Expand All @@ -22,14 +23,17 @@ go_library(
go_test(
name = "go_default_test",
srcs = [
"consumer_internal_test.go",
"consumer_test.go",
"registry_test.go",
],
embed = [":go_default_library"],
deps = [
":go_default_library",
"//platform/base/messagequeue:go_default_library",
"//platform/consumer/mock:go_default_library",
"//platform/errs:go_default_library",
"//platform/extension/consumergate:go_default_library",
"//platform/extension/consumergate/noop:go_default_library",
"//platform/extension/messagequeue:go_default_library",
"//platform/extension/messagequeue/mock:go_default_library",
"//submitqueue/core/topickey:go_default_library",
Expand Down
207 changes: 201 additions & 6 deletions platform/consumer/consumer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (

"github.com/uber-go/tally"
"github.com/uber/submitqueue/platform/errs"
"github.com/uber/submitqueue/platform/extension/consumergate"
extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
"github.com/uber/submitqueue/platform/metrics"
"go.uber.org/zap"
Expand All @@ -32,6 +33,16 @@ const (
// startupCleanupTimeoutMs is the timeout for cleaning up subscriptions when
// a controller fails to start during Start().
startupCleanupTimeoutMs = 30000

// gateExtensionMs is the visibility extension applied to a delivery blocked
// behind its consumer gate on each keep-in-flight tick, keeping it in-flight
// without burning retry budget (milliseconds). Must comfortably exceed
// defaultGateExtendInterval.
gateExtensionMs = int64(30000)

// defaultGateExtendInterval is how often a gate-blocked delivery's
// visibility is extended.
defaultGateExtendInterval = 10 * time.Second
)

// Consumer orchestrates multiple queue consumers. It handles subscription lifecycle,
Expand Down Expand Up @@ -61,6 +72,12 @@ type consumer struct {
metricsScope tally.Scope
registry TopicRegistry
processor errs.ErrorProcessor
gate consumergate.Gate

// gateExtendInterval is how often a gate-blocked delivery's visibility is
// extended. Fixed to defaultGateExtendInterval by New; a field (not the
// const) so in-package tests can exercise the keep-in-flight path quickly.
gateExtendInterval time.Duration

mu sync.Mutex
stopped bool
Expand All @@ -85,13 +102,20 @@ type activeSubscription struct {
// consumers such as DLQ reconciliation that must redeliver on any failure.
// processor must not be nil; callers that genuinely want no transformation
// can pass errs.NewClassifierProcessor() with no classifiers.
func New(logger *zap.SugaredLogger, scope tally.Scope, registry TopicRegistry, processor errs.ErrorProcessor) Consumer {
//
// gate is the consumer-gate implementation consulted before each delivery
// reaches its controller. Pass noop.New() (from
// platform/extension/consumergate/noop) for services that do not need runtime
// gating. gate must not be nil.
func New(logger *zap.SugaredLogger, scope tally.Scope, registry TopicRegistry, processor errs.ErrorProcessor, gate consumergate.Gate) Consumer {
return &consumer{
logger: logger,
metricsScope: scope.SubScope("consumer"),
registry: registry,
processor: processor,
subscriptions: make(map[TopicKey]*activeSubscription),
logger: logger,
metricsScope: scope.SubScope("consumer"),
registry: registry,
processor: processor,
gate: gate,
gateExtendInterval: defaultGateExtendInterval,
subscriptions: make(map[TopicKey]*activeSubscription),
}
}

Expand Down Expand Up @@ -341,6 +365,14 @@ func (m *consumer) processPartition(ctx context.Context, controller Controller,
func (m *consumer) processDelivery(ctx context.Context, controller Controller, delivery extqueue.Delivery, controllerScope tally.Scope) {
const opName = "process"

// Consumer gate: block the delivery while the controller's gate is closed.
// A false return means the consumer is shutting down while blocked — leave
// the delivery in-flight (no process, no ack/nack) so its visibility lapses
// into a normal redelivery. Gate errors fail open inside waitGate.
if !m.waitGate(ctx, controller, delivery, controllerScope) {
return
}

start := time.Now()
metrics.NamedCounter(controllerScope, opName, "messages_received", 1)

Expand Down Expand Up @@ -485,6 +517,169 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
)
}

// waitGate clears a delivery through the consumer gate before it reaches the
// controller. It returns true when the delivery may proceed, false when it
// must be dropped without processing or ack/nack (the visibility timeout then
// lapses into a normal redelivery).
//
// Gate.Enter checks the gate synchronously; an unblocked entry is the common
// path and costs nothing further. Only for a blocked entry does the consumer
// arrange its queue mechanics: the entry wait and a keep-in-flight extender
// run as goroutines under their own child contexts, supervised by this
// routine, which cancels and joins both before the delivery proceeds or is
// dropped. Failures fail open: if gate state cannot be read or recorded, or
// the delivery can no longer be held safely because a visibility extension
// failed, the delivery proceeds and the failure is surfaced via log and
// counter. Only consumer shutdown drops the delivery.
func (m *consumer) waitGate(ctx context.Context, controller Controller, delivery extqueue.Delivery, scope tally.Scope) bool {
const opName = "gate"

msg := delivery.Message()
consumerGroup := controller.ConsumerGroup()
topic := controller.TopicKey().String()

entry, err := m.gate.Enter(ctx, consumergate.Key{ConsumerGroup: consumerGroup, PartitionKey: msg.PartitionKey})
if err != nil {
// Gate state could not be read: fail open — gating is auxiliary, and
// a broken gate medium must not become a pipeline stall.
metrics.NamedCounter(scope, opName, "enter_errors", 1)
m.logger.Errorw("gate check failed, failing open",
"consumer_group", consumerGroup,
"topic", topic,
"message_id", msg.ID,
"error", err,
)
return true
}
if !entry.Blocked() {
return true
}

// The delivery is blocked; only now are the parked descriptor built and
// the queue mechanics arranged. The main routine supervises two
// goroutines — the entry wait and the keep-in-flight extender — each
// under its own child context, and reacts to whichever finishes first.
parked := consumergate.Parked{
Topic: topic,
MessageID: msg.ID,
Payload: msg.Payload,
Attempt: delivery.Attempt(),
}
waitCtx, cancelWait := context.WithCancel(ctx)
defer cancelWait()
extendCtx, cancelExtend := context.WithCancel(ctx)

extendErrCh, extendDone := m.keepInFlight(extendCtx, delivery)
// Whatever path exits, cancel the extender and join it, so no extension
// races the controller (or the redelivery, when the delivery is dropped).
defer func() {
cancelExtend()
<-extendDone
}()

start := time.Now()
defer func() {
metrics.NamedLatencyHistogram(scope, opName, "wait_latency", time.Since(start))
}()
waitResult := make(chan error, 1)
go func() {
waitResult <- entry.Wait(waitCtx, parked)
}()

select {
case err = <-waitResult:
// The wait finished on its own; fall through to classify its result.
case extendErr := <-extendErrCh:
// The delivery can no longer be held safely (its visibility may lapse
// and the queue redeliver it): cancel the wait and join it, then fail
// open — unless the consumer is shutting down anyway, in which case
// the delivery is dropped for redelivery like any other shutdown.
cancelWait()
<-waitResult
if ctx.Err() != nil {
metrics.NamedCounter(scope, opName, "shutdown_while_blocked", 1)
m.logger.Infow("consumer shutdown while delivery was blocked by gate",
"consumer_group", consumerGroup,
"topic", topic,
"message_id", msg.ID,
)
return false
}
metrics.NamedCounter(scope, opName, "extend_errors", 1)
m.logger.Errorw("failed to keep gate-blocked delivery in-flight, failing open",
"consumer_group", consumerGroup,
"topic", topic,
"message_id", msg.ID,
"error", extendErr,
)
return true
}

if err == nil {
return true
}

if errors.Is(err, context.Canceled) {
// On this path the wait context was never cancelled by the extend
// branch above, so a cancelled wait means the parent was cancelled:
// the consumer is shutting down and the delivery is dropped without
// processing or ack/nack.
metrics.NamedCounter(scope, opName, "shutdown_while_blocked", 1)
m.logger.Infow("consumer shutdown while delivery was blocked by gate",
"consumer_group", consumerGroup,
"topic", topic,
"message_id", msg.ID,
)
return false
}

// Gate state could not be re-read or the record could not be written:
// fail open, as above.
metrics.NamedCounter(scope, opName, "wait_errors", 1)
m.logger.Errorw("gate wait failed, failing open",
"consumer_group", consumerGroup,
"topic", topic,
"message_id", msg.ID,
"error", err,
)
return true
}

// keepInFlight starts a goroutine that periodically extends the delivery's
// visibility so the queue does not redeliver a gate-blocked message. The
// goroutine's lifecycle is controlled by ctx: it extends on each tick until
// ctx is cancelled or an extension fails. A failure means the delivery can no
// longer be held safely (its visibility may lapse and the queue redeliver
// it); it is sent on the returned error channel for the supervising routine
// to act on, and ends the goroutine.
//
// The returned done channel closes when the goroutine exits; callers cancel
// ctx and receive from done to join the goroutine before letting the delivery
// proceed, guaranteeing no extension races the controller.
func (m *consumer) keepInFlight(ctx context.Context, delivery extqueue.Delivery) (extendErr <-chan error, done <-chan struct{}) {
errCh := make(chan error, 1)
doneCh := make(chan struct{})

go func() {
defer close(doneCh)
ticker := time.NewTicker(m.gateExtendInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
}
if err := delivery.ExtendVisibilityTimeout(ctx, gateExtensionMs); err != nil {
errCh <- fmt.Errorf("failed to extend visibility of gate-blocked delivery: %w", err)
return
}
}
}()

return errCh, doneCh
}

// Stop gracefully shuts down all handlers with the specified timeout.
// Cancels all subscription contexts and waits for consumption goroutines to finish.
// timeoutMs is the maximum time in milliseconds to wait for graceful shutdown.
Expand Down
Loading