@@ -23,6 +23,7 @@ import (
2323
2424 "github.com/uber-go/tally"
2525 "github.com/uber/submitqueue/platform/errs"
26+ "github.com/uber/submitqueue/platform/extension/consumergate"
2627 extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
2728 "github.com/uber/submitqueue/platform/metrics"
2829 "go.uber.org/zap"
@@ -32,6 +33,16 @@ const (
3233 // startupCleanupTimeoutMs is the timeout for cleaning up subscriptions when
3334 // a controller fails to start during Start().
3435 startupCleanupTimeoutMs = 30000
36+
37+ // gateExtensionMs is the visibility extension applied to a delivery blocked
38+ // behind its consumer gate on each keep-in-flight tick, keeping it in-flight
39+ // without burning retry budget (milliseconds). Must comfortably exceed
40+ // defaultGateExtendInterval.
41+ gateExtensionMs = int64 (30000 )
42+
43+ // defaultGateExtendInterval is how often a gate-blocked delivery's
44+ // visibility is extended.
45+ defaultGateExtendInterval = 10 * time .Second
3546)
3647
3748// Consumer orchestrates multiple queue consumers. It handles subscription lifecycle,
@@ -61,6 +72,12 @@ type consumer struct {
6172 metricsScope tally.Scope
6273 registry TopicRegistry
6374 processor errs.ErrorProcessor
75+ gate consumergate.Gate
76+
77+ // gateExtendInterval is how often a gate-blocked delivery's visibility is
78+ // extended. Fixed to defaultGateExtendInterval by New; a field (not the
79+ // const) so in-package tests can exercise the keep-in-flight path quickly.
80+ gateExtendInterval time.Duration
6481
6582 mu sync.Mutex
6683 stopped bool
@@ -87,13 +104,20 @@ type activeSubscription struct {
87104// without introducing duplicate consumer sub-scopes. processor must not be nil;
88105// callers that genuinely want no transformation can pass
89106// errs.NewClassifierProcessor() with no classifiers.
90- func New (logger * zap.SugaredLogger , scope tally.Scope , registry TopicRegistry , processor errs.ErrorProcessor ) Consumer {
107+ //
108+ // gate is the consumer-gate implementation consulted before each delivery
109+ // reaches its controller. Pass noop.New() (from
110+ // platform/extension/consumergate/noop) for services that do not need runtime
111+ // gating. gate must not be nil.
112+ func New (logger * zap.SugaredLogger , scope tally.Scope , registry TopicRegistry , processor errs.ErrorProcessor , gate consumergate.Gate ) Consumer {
91113 return & consumer {
92- logger : logger ,
93- metricsScope : scope ,
94- registry : registry ,
95- processor : processor ,
96- subscriptions : make (map [TopicKey ]* activeSubscription ),
114+ logger : logger ,
115+ metricsScope : scope ,
116+ registry : registry ,
117+ processor : processor ,
118+ gate : gate ,
119+ gateExtendInterval : defaultGateExtendInterval ,
120+ subscriptions : make (map [TopicKey ]* activeSubscription ),
97121 }
98122}
99123
@@ -343,6 +367,14 @@ func (m *consumer) processPartition(ctx context.Context, controller Controller,
343367func (m * consumer ) processDelivery (ctx context.Context , controller Controller , delivery extqueue.Delivery , controllerScope tally.Scope ) {
344368 const opName = "process"
345369
370+ // Consumer gate: block the delivery while the controller's gate is closed.
371+ // A false return means the consumer is shutting down while blocked — leave
372+ // the delivery in-flight (no process, no ack/nack) so its visibility lapses
373+ // into a normal redelivery. Gate errors fail open inside waitGate.
374+ if ! m .waitGate (ctx , controller , delivery , controllerScope ) {
375+ return
376+ }
377+
346378 msg := delivery .Message ()
347379 topicKey := controller .TopicKey ()
348380
@@ -466,6 +498,103 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
466498 )
467499}
468500
501+ // waitGate clears a delivery through the consumer gate before it reaches the
502+ // controller. It returns true when the delivery may proceed, false when it
503+ // must be left in-flight without processing or ack/nack.
504+ //
505+ // Gate.Enter checks the gate synchronously; an unblocked entry is the common
506+ // path and costs nothing further. For a blocked entry the gate hands back a
507+ // watch channel (its own monitoring goroutine behind it), and this routine
508+ // multiplexes the watch with visibility extension. The source delivery remains
509+ // owned by the queue throughout the wait: gating never acknowledges, rejects,
510+ // nacks, or moves it. On shutdown, extension stops and normal queue visibility
511+ // semantics make the delivery eligible for redelivery.
512+ //
513+ // Failures fail open: if gate state cannot be read or recorded, or the delivery
514+ // can no longer be held safely because visibility extension failed, processing
515+ // proceeds and the failure is surfaced via logs and metrics.
516+ func (m * consumer ) waitGate (ctx context.Context , controller Controller , delivery extqueue.Delivery , scope tally.Scope ) bool {
517+ const opName = "gate"
518+
519+ msg := delivery .Message ()
520+ consumerGroup := controller .ConsumerGroup ()
521+ topic := controller .TopicKey ().String ()
522+
523+ entry , err := m .gate .Enter (ctx , consumergate.Key {ConsumerGroup : consumerGroup , PartitionKey : msg .PartitionKey })
524+ if err != nil {
525+ if errors .Is (err , context .Canceled ) {
526+ return false
527+ }
528+ metrics .NamedCounter (scope , opName , "enter_errors" , 1 )
529+ m .logger .Errorw ("gate check failed, failing open" ,
530+ "consumer_group" , consumerGroup ,
531+ "topic" , topic ,
532+ "message_id" , msg .ID ,
533+ "error" , err ,
534+ )
535+ return true
536+ }
537+ if ! entry .Blocked () {
538+ return true
539+ }
540+
541+ start := time .Now ()
542+ defer func () {
543+ metrics .NamedHistogram (scope , opName , "wait_latency" , metrics .LongLatencyBuckets ).RecordDuration (time .Since (start ))
544+ }()
545+
546+ descriptor := consumergate.DeliveryDescriptor {
547+ Topic : topic ,
548+ MessageID : msg .ID ,
549+ Payload : msg .Payload ,
550+ Attempt : delivery .Attempt (),
551+ }
552+
553+ watchCtx , cancelWatch := context .WithCancel (ctx )
554+ defer cancelWatch ()
555+ watchCh := entry .Watch (watchCtx , descriptor )
556+
557+ ticker := time .NewTicker (m .gateExtendInterval )
558+ defer ticker .Stop ()
559+
560+ for {
561+ select {
562+ case waitErr := <- watchCh :
563+ if waitErr == nil {
564+ return true
565+ }
566+ if errors .Is (waitErr , context .Canceled ) {
567+ return false
568+ }
569+ metrics .NamedCounter (scope , opName , "wait_errors" , 1 )
570+ m .logger .Errorw ("gate wait failed, failing open" ,
571+ "consumer_group" , consumerGroup ,
572+ "topic" , topic ,
573+ "message_id" , msg .ID ,
574+ "error" , waitErr ,
575+ )
576+ return true
577+
578+ case <- ticker .C :
579+ if extendErr := delivery .ExtendVisibilityTimeout (ctx , gateExtensionMs ); extendErr != nil {
580+ cancelWatch ()
581+ <- watchCh
582+ if errors .Is (extendErr , context .Canceled ) {
583+ return false
584+ }
585+ metrics .NamedCounter (scope , opName , "wait_errors" , 1 )
586+ m .logger .Errorw ("gate visibility extension failed, failing open" ,
587+ "consumer_group" , consumerGroup ,
588+ "topic" , topic ,
589+ "message_id" , msg .ID ,
590+ "error" , extendErr ,
591+ )
592+ return true
593+ }
594+ }
595+ }
596+ }
597+
469598func controllerClassificationTags (err error ) []metrics.Tag {
470599 origin := "infra"
471600 if errs .IsRetryable (err ) {
0 commit comments