fix(metrics): count gate_closed where budget gating actually happens

README.md

@@ -574,7 +574,7 @@ The Async Processor exposes Prometheus metrics under the `llm_d_async` subsystem
 | `llm_d_async_async_queue_residence_time_millis` | Histogram | Time in milliseconds a message spent buffered in-process, from broker ingestion until a worker pulled it for processing. Measures the async delay introduced by the system (queue time). Always registered. |
 | `llm_d_async_async_dispatch_budget` | Gauge | Current dispatch budget [0.0–1.0] returned by the queue's gate; the fraction of system capacity available for new requests (0.0 = gate fully closed). Useful for diagnosing why throughput is throttled. |
 | `llm_d_async_async_pool_worker_limit` | Gauge | Configured worker concurrency limit for a pool (carries only the `pool_name` label). Compare against `llm_d_async_async_inflight_requests` to compute worker utilization. |
-| `llm_d_async_async_gate_decisions_total` | Counter | Count of gate decisions that prevented a message from being dispatched, by `reason`: `gate_closed` (no dispatch budget), `quota_exhausted` (per-attribute quota overflow), `dropped` (gate permanently rejected the request), `error` (gate evaluation failed). |
+| `llm_d_async_async_gate_decisions_total` | Counter | Count of gate decisions that prevented dispatch, by `reason`: `gate_closed` (no dispatch budget), `quota_exhausted` (per-attribute quota overflow), `dropped` (gate permanently rejected the request), `error` (gate evaluation failed). `quota_exhausted`, `dropped` and `error` count individual messages refused after being dequeued. `gate_closed` counts those plus every dequeue round in which the budget shrank the batch to zero — the way budget-based gates (`prometheus-budget`/`-saturation`/`-query`) shed work *before* a message is dequeued — so its rate reflects throttled dispatch opportunities, not messages. All four `reason` series are created at 0 when a queue starts, so a query returns 0 rather than an empty vector. |
 | `llm_d_async_async_gate_metric_value` | Gauge | Raw metric value a metric-based gate (`prometheus-saturation`/`-budget`/`-query`) last read — the number compared against the threshold below. For the saturation gate this is `1 - saturation`. |
 | `llm_d_async_async_gate_metric_threshold` | Gauge | Threshold the value above is compared against. The gate closes when `value <= threshold`, which is what drives `async_dispatch_budget` to 0. |
 

pkg/metrics/metrics.go

@@ -112,7 +112,7 @@ var (
 	}, []string{LabelPoolName})
 	GateDecisions = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Subsystem: SchedulerSubsystem, Name: "async_gate_decisions_total",
-		Help: "Count of gate decisions that prevented a message from being dispatched, by reason (gate_closed, quota_exhausted, dropped, error).",
+		Help: "Count of gate decisions that prevented dispatch, by reason (gate_closed, quota_exhausted, dropped, error). quota_exhausted/dropped/error count individual messages refused after they were dequeued; gate_closed counts those plus each dequeue round in which the gate's budget shrank the batch to zero, which is how budget-based gates (prometheus-budget/-saturation/-query) shed work before any message is dequeued.",
 	}, []string{LabelQueueID, LabelQueueName, LabelPoolName, LabelReason})
 	GateMetricValue = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 		Subsystem: SchedulerSubsystem, Name: "async_gate_metric_value",
@@ -212,6 +212,17 @@ func RecordGateDecision(reason, queueID, queueName, poolName string) {
 	GateDecisions.WithLabelValues(queueID, queueName, poolName, reason).Inc()
 }
 
+// InitGateDecisions pre-creates a queue's async_gate_decisions_total series with
+// every reason at 0. A CounterVec label set that has never been incremented is
+// absent from /metrics entirely, so querying a reason that has not fired yet
+// yields an empty vector rather than 0 — indistinguishable from a queue that was
+// never configured or a scrape that never landed.
+func InitGateDecisions(queueID, queueName, poolName string) {
+	for _, reason := range []string{ReasonGateClosed, ReasonQuotaExhausted, ReasonDropped, ReasonError} {
+		GateDecisions.WithLabelValues(queueID, queueName, poolName, reason)
+	}
+}
+
 // SetGateMetricValue records the raw metric value a metric-based dispatch gate
 // last read and the threshold it is compared against. Helps answer "why is the
 // gate closed?" (value <= threshold). queueID/queueName/poolName identify the

pkg/metrics/metrics_test.go

@@ -89,6 +89,21 @@ func TestRecordGateDecision(t *testing.T) {
 	}
 }
 
+func TestInitGateDecisions(t *testing.T) {
+	InitGateDecisions("q10", "queue-10", "pool-y")
+
+	for _, reason := range []string{ReasonGateClosed, ReasonQuotaExhausted, ReasonDropped, ReasonError} {
+		// A never-incremented CounterVec label set is absent from /metrics, so
+		// the point of the pre-creation is that these series exist and read 0.
+		if got := testutil.ToFloat64(GateDecisions.WithLabelValues("q10", "queue-10", "pool-y", reason)); got != 0 {
+			t.Errorf("GateDecisions[%s] = %v, want 0", reason, got)
+		}
+	}
+	if got := testutil.CollectAndCount(GateDecisions, "llm_d_async_async_gate_decisions_total"); got < 4 {
+		t.Errorf("GateDecisions series count = %d, want at least 4", got)
+	}
+}
+
 func TestGetAsyncProcessorCollectors_includesGateDecisions(t *testing.T) {
 	for _, withLatency := range []bool{false, true} {
 		collectors := GetAsyncProcessorCollectors(withLatency)

pkg/pubsub/pubsubimpl.go

@@ -523,6 +523,8 @@ func (r *PubSubMQFlow) requestWorker(ctx context.Context, pubSubClient *pubsub.C
 
 	sub := pubSubClient.Subscriber(subscriberID)
 
+	metrics.InitGateDecisions("", subscriberID, poolID)
+
 	for ctx.Err() == nil {
 		receiveCtx, cancel := context.WithCancel(ctx)
 		budget := gate.Budget(ctx)
@@ -549,6 +551,13 @@ func (r *PubSubMQFlow) requestWorker(ctx context.Context, pubSubClient *pubsub.C
 		sub.ReceiveSettings.MaxOutstandingMessages = currBatchSize
 		sub.ReceiveSettings.NumGoroutines = 1
 		if currBatchSize <= 0 {
+			// Same pre-dequeue back-pressure as the sorted-set path: with no
+			// outstanding-message slots the receive callback never runs, so
+			// gate.Apply never records the refusal. Count the throttled receive
+			// window instead (#368). Unlike Redis there is no cheap depth probe
+			// — the subscription backlog comes from Cloud Monitoring — so this
+			// counts the window whether or not messages happen to be waiting.
+			metrics.RecordGateDecision(metrics.ReasonGateClosed, "", subscriberID, poolID)
 			<-receiveCtx.Done()
 			cancel()
 			continue

pkg/redis/sortedset_impl.go

@@ -418,6 +418,8 @@ func (r *RedisSortedSetFlow) requestWorker(ctx context.Context, msgChannel chan
 		gate = r.gate
 	}
 
+	metrics.InitGateDecisions(queueID, queueName, r.poolNameFor(queueID))
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -428,16 +430,36 @@ func (r *RedisSortedSetFlow) requestWorker(ctx context.Context, msgChannel chan
 	}
 }
 
+// poolNameFor returns the worker pool a queue routes to, or "" when the queue has
+// no config entry (the metric label is then empty rather than wrong).
+func (r *RedisSortedSetFlow) poolNameFor(queueID string) string {
+	if cfg, ok := r.configMap[queueID]; ok {
+		return cfg.WorkerPoolID
+	}
+	return ""
+}
+
 func (r *RedisSortedSetFlow) processMessages(ctx context.Context, msgChannel chan *api.InternalRequest, queueName string, queueID string, gate pipeline.Gate, logger logr.Logger) {
 	currentTime := float64(time.Now().Unix())
 
 	budget := gate.Budget(ctx)
-	poolName := ""
-	if cfg, ok := r.configMap[queueID]; ok {
-		poolName = cfg.WorkerPoolID
-	}
+	poolName := r.poolNameFor(queueID)
 	metrics.SetDispatchBudget(budget, queueID, queueName, poolName)
 	batchSize := int(math.Floor(float64(r.batchSize) * budget))
+	if batchSize <= 0 {
+		// Back-pressure here is applied pre-dequeue: the budget shrank the batch
+		// to zero, so no message reaches gate.Apply below — the only other site
+		// that records a refusal. Count the throttled poll itself, or gate_closed
+		// stays silent exactly while the gate is doing its job (#368). Only count
+		// when work is actually waiting; an idle queue was not held back.
+		depth, err := r.rdb.ZCard(ctx, queueName).Result()
+		if err != nil {
+			logger.V(logutil.DEFAULT).Error(err, "Failed to read queue depth for a closed gate", "queue", queueName)
+		} else if depth > 0 {
+			metrics.RecordGateDecision(metrics.ReasonGateClosed, queueID, queueName, poolName)
+		}
+		return
+	}
 
 	for i := 0; i < batchSize; i++ {
 		results, err := r.rdb.ZPopMin(ctx, queueName, 1).Result()

pkg/redis/sortedset_impl_test.go

@@ -12,8 +12,11 @@ import (
 	"time"
 
 	"github.com/alicebob/miniredis/v2"
+	"github.com/go-logr/logr"
 	"github.com/llm-d/llm-d-async/api"
 	"github.com/llm-d/llm-d-async/pipeline"
+	"github.com/llm-d/llm-d-async/pkg/metrics"
+	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/redis/go-redis/v9"
 )
 
@@ -1037,6 +1040,55 @@ func TestSortedSetFlow_ZeroBudget(t *testing.T) {
 	}
 }
 
+func TestSortedSetFlow_ClosedGateRecordsGateClosedDecision(t *testing.T) {
+	s, rdb, ctx, cancel := setupTest(t)
+	defer s.Close()
+	defer rdb.Close() // nolint:errcheck
+	defer cancel()
+
+	queue := "gate-closed-metric-queue"
+	queueID := "gate-closed-metric-id"
+	gateClosed := func() float64 {
+		return testutil.ToFloat64(metrics.GateDecisions.WithLabelValues(queueID, queue, "", metrics.ReasonGateClosed))
+	}
+
+	flow := &RedisSortedSetFlow{
+		rdb:          rdb,
+		pollInterval: 50 * time.Millisecond,
+		batchSize:    10,
+		gate:         pipeline.DispatchGateFunc(func(ctx context.Context) float64 { return 0.0 }),
+	}
+	msgChannel := make(chan *api.InternalRequest, 1)
+
+	// Nothing queued: the gate held nothing back, so nothing is counted.
+	flow.processMessages(ctx, msgChannel, queue, queueID, flow.gate, logr.Discard())
+	if got := gateClosed(); got != 0 {
+		t.Fatalf("gate_closed on an empty queue = %v, want 0", got)
+	}
+
+	msg := api.RequestMessage{
+		ID:       "gate-closed-1",
+		Created:  time.Now().Unix(),
+		Deadline: 9999999999,
+		Payload:  map[string]any{"test": "data"},
+	}
+	rdb.ZAdd(ctx, queue, redis.Z{Score: float64(time.Now().Unix()), Member: envelopeJSON(msg)})
+
+	// Work is waiting and the budget zeroes the batch: each throttled poll is a
+	// gate_closed decision, even though no message is ever dequeued (#368).
+	flow.processMessages(ctx, msgChannel, queue, queueID, flow.gate, logr.Discard())
+	flow.processMessages(ctx, msgChannel, queue, queueID, flow.gate, logr.Discard())
+	if got := gateClosed(); got != 2 {
+		t.Fatalf("gate_closed with a backlogged queue = %v, want 2", got)
+	}
+	if len(msgChannel) != 0 {
+		t.Fatalf("message dispatched while the gate was closed")
+	}
+	if count, _ := rdb.ZCard(ctx, queue).Result(); count != 1 {
+		t.Fatalf("queue depth = %d, want the message left in place", count)
+	}
+}
+
 func TestSortedSetFlow_ResultRetryAfterFailure(t *testing.T) {
 	s, rdb, ctx, cancel := setupTest(t)
 	defer s.Close()

release-notes.d/unreleased/371.md

@@ -0,0 +1,7 @@
+---
+pr: 371
+url: https://github.com/llm-d/llm-d-async/pull/371
+author: shimib
+date: 2026-07-28
+---
+Fix `async_gate_decisions_total{reason="gate_closed"}`, which could never increment for a budget-based gate: back-pressure is applied pre-dequeue by shrinking the dispatch batch to zero, while the counter was only incremented post-dequeue inside the loop that batch size had just emptied — and a counter series that never increments is absent from `/metrics` entirely, so the query returned an empty vector rather than 0. The decision is now recorded where it is made: the sorted-set path counts each throttled poll whose batch the budget zeroed while the queue is non-empty, and the Pub/Sub path counts each receive window it skips for the same reason. Both backends also pre-create all four `reason` series at 0 when a queue starts. The counter's unit is decisions, not messages; the help text and README metric table now say so.