fix: tierpriority catch-all, helm dead flags, SS gating cleanup, pubsub buffering

P1 — tierpriority strands unknown labels
popNext only polled buckets in the configured ClassOrder × TierOrder
(plus empty-string fallbacks). A message with tier=urgent or class=gold
(typo or schema drift) landed in a bucket nothing consults — work
stranded until the input channel closed.

Added popUnconfigured: a final catch-all pass that drains any bucket
whose (tier, class) key isn't covered by the configured order, at
lowest priority. The policy is now total over its input domain.

P1 — helm chart emits removed flags
ap-deployments.yaml still emitted --redis.ss.gate-type and
--redis.ss.gate-params (flags removed when the SS gate impl was
dropped). Any operator setting ap.redis.gateType would crashloop on
"flag provided but not defined". Removed the block, the matching
values.yaml defaults, the dead gateParamsJson helper, and the e2e
deploy values references.

P1 — SS gating landing-state inconsistency
WithGateFactory on RedisSortedSetFlow was stored-but-unread, main.go
logged "with per-queue gating", and README told users to switch to SS
for per-queue gating — all false (SS gates aren't wired). Removed
WithGateFactory + SortedSetOption + gateFactory field, dropped the
arg from main.go, fixed the log line, and corrected README to direct
operators at gcp-pubsub-gated for actual per-queue gating.

P2 — pubsub callback could block prefetch under saturation
The per-subscription channel was unbuffered, so under sustained
saturation a stalled merge goroutine propagated straight back to the
receive callback, halting Pub/Sub prefetch. Worse, ctx cancellation
mid-stall deadlocked the callback (Receive() never exits).

Buffered the channel to the same prefetch depth as Pub/Sub's
MaxOutstandingMessages (extracted prefetchBufferDepth helper) so the
callback is non-blocking up to the prefetch window. Wrapped the send
in select-with-ctx so shutdown unblocks and any subscription-gate
releases fire cleanly on cancel.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: joeltg <joel@reflection.ai>

Author: joeltg

README.md

@@ -280,7 +280,7 @@ The configuration file when using the `redis.queues-config-file` flag should hav
 ]
 ```
 
-<u>Note:</u> The ephemeral Redis Channels implementation does not support per-queue dispatch gates. Use the [Redis Sorted Set](#redis-sorted-set-persisted) implementation for per-queue gating.
+<u>Note:</u> Per-queue / per-pool dispatch gates are currently only wired on the `gcp-pubsub-gated` flow. The Redis Channels and Redis Sorted Set flows do not yet apply gate chains.
 
 **Configuration Fields:**
 

charts/async-processor/templates/_helpers.tpl

@@ -57,18 +57,6 @@ Create the name of the service account to use
 {{- default (include "async-processor.fullname" .) .Values.serviceAccount.name }}
 {{- end }}
 
-{{/*
-Render gate params as JSON with all values as strings.
-The gate params parser expects map[string]string, so numeric values must be quoted.
-*/}}
-{{- define "async-processor.gateParamsJson" -}}
-{{- $out := dict -}}
-{{- range $k, $v := .Values.ap.redis.gateParams -}}
-{{- $_ := set $out $k ($v | toString) -}}
-{{- end -}}
-{{- $out | toJson -}}
-{{- end }}
-
 {{/*
 Resolve the Redis secret name.
 If redis.url is set, the chart creates a Secret named <fullname>-redis.

charts/async-processor/templates/ap-deployments.yaml

@@ -37,10 +37,6 @@ spec:
           - "{{ .Values.ap.redis.pollIntervalMs | default 1000 }}"
           - --redis.ss.batch-size
           - "{{ .Values.ap.redis.batchSize | default 10 }}"
-          {{- if .Values.ap.redis.gateType }}
-          - --redis.ss.gate-type={{ .Values.ap.redis.gateType }}
-          - --redis.ss.gate-params={{ include "async-processor.gateParamsJson" . }}
-          {{- end }}
           {{- else }}
           - --message-queue-impl=redis-pubsub
           - --redis.igw-base-url={{ .Values.ap.igwBaseURL }}

charts/async-processor/values.yaml

@@ -38,13 +38,12 @@ ap:
     resultQueueName: "result-list"
     pollIntervalMs: 1000
     batchSize: 10
-    gateType: ""
-    gateParams: {}
+    # Note: per-pool / per-subscription gate chains are configured via
+    # the topics-config-file (see README), not chart values. Chart-level
+    # support for the new gate types is not yet wired.
   # PodMonitor for model server metrics relabeling.
   # Creates a PodMonitor that scrapes vLLM metrics and relabels
   # the inference_pool pod label into scraped metrics.
-  # Required when using prometheus-budget or prometheus-saturation gates
-  # without llm-d's flow control plugin enabled.
   modelServerMonitor:
     enabled: false
     selector:

cmd/main.go

@@ -122,13 +122,13 @@ func main() {
 		}
 		impl = flow
 	case "redis-sortedset":
-		flow, err := redis.NewRedisSortedSetFlow(redis.WithGateFactory(gateFactory))
+		flow, err := redis.NewRedisSortedSetFlow()
 		if err != nil {
 			setupLog.Error(err, "Failed to create Redis sorted-set flow")
 			os.Exit(1)
 		}
 		impl = flow
-		setupLog.Info("Using Redis sorted-set flow with per-queue gating")
+		setupLog.Info("Using Redis sorted-set flow")
 	case "gcp-pubsub":
 		impl = pubsub.NewGCPPubSubMQFlow()
 	case "gcp-pubsub-gated":

docs/guides/e2e-deploy/async-processor-values.yaml

@@ -19,11 +19,6 @@ ap:
     resultQueueName: "result-list"
     pollIntervalMs: 1000
     batchSize: 10
-    gateType: "prometheus-budget"
-    gateParams:
-      pool: "optimized-baseline"
-      max_concurrency: "100"
-      baseline: "0.05"
   modelServerMonitor:
     enabled: true
     selector:

pkg/async/inference/mergepolicy/tierpriority/policy.go

@@ -16,6 +16,11 @@
 // pipeline knows nothing about them. Operators configure the label key
 // and value-ordering, so different vocabularies (or more than two
 // classes) drop in without code changes.
+//
+// Messages whose tier or class label values aren't in the configured
+// order (operator typos, schema drift) aren't lost: they drain at
+// lowest priority via a catch-all pass after the configured priority
+// passes complete. The policy is total over its input domain.
 package tierpriority
 
 import (
@@ -106,6 +111,32 @@ type Policy struct {
 	cfg Config
 }
 
+// isConfiguredKey reports whether (tier, class) is one of the buckets
+// popNext drains in its configured passes. Used by popUnconfigured to
+// identify catch-all candidates. Empty-string values count as
+// configured (the configured passes consult them explicitly).
+func (p *Policy) isConfiguredKey(k bucketKey) bool {
+	tierOK := k.tier == ""
+	if !tierOK {
+		for _, t := range p.cfg.TierOrder {
+			if t == k.tier {
+				tierOK = true
+				break
+			}
+		}
+	}
+	classOK := k.class == ""
+	if !classOK {
+		for _, c := range p.cfg.ClassOrder {
+			if c == k.class {
+				classOK = true
+				break
+			}
+		}
+	}
+	return tierOK && classOK
+}
+
 var _ pipeline.RequestMergePolicy = (*Policy)(nil)
 
 // MergeRequestChannels groups incoming RequestChannels by pool, spawns
@@ -248,13 +279,40 @@ func (ps *poolState) popNext() *pipeline.EmbelishedRequestMessage {
 		if m := ps.popBucket(bucketKey{"", ""}); m != nil {
 			return m
 		}
+		// Catch-all: any bucket whose (tier, class) isn't covered by
+		// the configured ClassOrder / TierOrder still contains real
+		// work. Drain it at lowest priority — keeps the policy total
+		// over its input domain so typos or schema drift (e.g.
+		// `tier=urgent` against a config that only declares
+		// interactive/async/batch) don't strand messages.
+		if m := ps.popUnconfigured(); m != nil {
+			return m
+		}
 		if ps.closed {
 			return nil
 		}
 		ps.cond.Wait()
 	}
 }
 
+// popUnconfigured pops one message from any bucket whose (tier, class)
+// key is not in the configured ClassOrder / TierOrder. The configured
+// passes above already drained their buckets; whatever remains is, by
+// definition, unconfigured. Iteration order is map-iteration (random)
+// — these are operator-misconfigured messages, so we don't promise any
+// particular interleaving among them.
+func (ps *poolState) popUnconfigured() *pipeline.EmbelishedRequestMessage {
+	for key := range ps.buckets {
+		if ps.policy.isConfiguredKey(key) {
+			continue
+		}
+		if m := ps.popBucket(key); m != nil {
+			return m
+		}
+	}
+	return nil
+}
+
 func (ps *poolState) popBucket(key bucketKey) *pipeline.EmbelishedRequestMessage {
 	b := ps.buckets[key]
 	if b == nil {

pkg/pubsub/pubsubimpl.go

@@ -398,7 +398,16 @@ func NewGCPPubSubMQFlow(opts ...PubSubOption) *PubSubMQFlow {
 			panic(fmt.Sprintf("failed to build subscription gates for %q: %v", cfg.SubscriberID, err))
 		}
 
-		ch := make(chan *pipeline.EmbelishedRequestMessage)
+		// Buffer the per-subscription channel to the prefetch depth so
+		// the receive callback's `ch <- emb` is non-blocking up to the
+		// prefetch capacity. Without this, a transient stall in the
+		// merge goroutine (e.g. pool output buffer full while workers
+		// drain a saturated backend) propagates straight back to the
+		// callback and halts Pub/Sub prefetch. Buffering decouples the
+		// two for burst absorption; sustained saturation past the
+		// buffer still applies backpressure to Pub/Sub, which is the
+		// correct signal at that point.
+		ch := make(chan *pipeline.EmbelishedRequestMessage, prefetchBufferDepth())
 		p.requestChannels = append(p.requestChannels, RequestChannelData{
 			requestChannel: pipeline.RequestChannel{
 				Channel:            ch,
@@ -596,10 +605,7 @@ func (r *PubSubMQFlow) requestWorker(ctx context.Context, pubSubClient *pubsub.C
 	//
 	// MaxOutstandingBytes = -1 removes the bytes-based limit (default
 	// 1GB) so only message count gates prefetch.
-	prefetchDepth := *batchSize * 5
-	if prefetchDepth < 1000 {
-		prefetchDepth = 1000
-	}
+	prefetchDepth := prefetchBufferDepth()
 	sub.ReceiveSettings.MaxOutstandingMessages = prefetchDepth
 	sub.ReceiveSettings.MaxOutstandingBytes = -1
 	sub.ReceiveSettings.NumGoroutines = 1
@@ -736,8 +742,19 @@ func (r *PubSubMQFlow) requestWorker(ctx context.Context, pubSubClient *pubsub.C
 			}
 		}
 
-		ch <- emb
-		stats.dispatched.Add(1)
+		// Forward to the merge policy. Under sustained saturation the
+		// buffered channel can still fill; select-with-ctx ensures
+		// shutdown unblocks rather than deadlocking the callback (and
+		// hence Receive(), which would never exit). On ctx cancel we
+		// fire any subscription-gate releases attached to emb and
+		// nack so Pub/Sub redelivers.
+		select {
+		case ch <- emb:
+			stats.dispatched.Add(1)
+		case <-ctx.Done():
+			emb.FireReleases()
+			resultsChannel <- ackAction{ack: false}
+		}
 	})
 	if err != nil {
 		logger.V(logutil.DEFAULT).Error(err, "Fail to receive messages from request subscription")
@@ -864,6 +881,19 @@ func synthesizePool(s TopicConfig) pipeline.Pool {
 	return pool
 }
 
+// prefetchBufferDepth is the shared sizing for both Pub/Sub's
+// MaxOutstandingMessages (transport-side prefetch) and the
+// per-subscription channel buffer (Go-side handoff to the merge
+// goroutine). Keeping them equal means the channel can hold the full
+// prefetch window without backpressuring into the callback.
+func prefetchBufferDepth() int {
+	depth := *batchSize * 5
+	if depth < 1000 {
+		depth = 1000
+	}
+	return depth
+}
+
 // bytesTrimLeftSpace returns b with leading ASCII whitespace removed.
 // Tiny helper to peek at the first non-whitespace byte without pulling
 // in the unicode/strings packages for one use site.

pkg/redis/sortedset_impl.go

@@ -56,22 +56,13 @@ type RedisSortedSetFlow struct {
 	resultChannel   chan api.ResultMessage
 	pollInterval    time.Duration
 	batchSize       int
-	gateFactory     pipeline.GateFactory
 }
 
-// SortedSetOption is a functional option for configuring RedisSortedSetFlow.
-type SortedSetOption func(*RedisSortedSetFlow)
-
-// WithGateFactory sets a GateFactory for subscription/pool gate
-// instantiation. Currently retained for forward compatibility; the
-// sortedset flow does not yet wire subscription or pool gate chains.
-func WithGateFactory(factory pipeline.GateFactory) SortedSetOption {
-	return func(r *RedisSortedSetFlow) {
-		r.gateFactory = factory
-	}
-}
-
-func NewRedisSortedSetFlow(opts ...SortedSetOption) (*RedisSortedSetFlow, error) {
+// Note: the sortedset flow does not currently wire subscription or
+// pool gate chains. Per-queue / per-pool gating is supported only on
+// the GCP Pub/Sub flow today (gcp-pubsub-gated). When SS gates are
+// re-introduced, a GateFactory will thread through here.
+func NewRedisSortedSetFlow() (*RedisSortedSetFlow, error) {
 	configs, err := loadQueueConfigs()
 	if err != nil {
 		return nil, err
@@ -89,10 +80,6 @@ func NewRedisSortedSetFlow(opts ...SortedSetOption) (*RedisSortedSetFlow, error)
 		batchSize:       *ssBatchSize,
 	}
 
-	for _, opt := range opts {
-		opt(r)
-	}
-
 	for _, cfg := range configs {
 		headers, err := util.ExpandEnvMapValues(cfg.HTTPHeaders)
 		if err != nil {