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

[shimib]

Fixes #368.

Problem

async_gate_decisions_total{reason="gate_closed"} is structurally pinned at zero for a budget-based gate. The two mechanisms are wired in series and the first starves the second:

// pkg/async/inference/flowcontrol/metric_dispatch_gate.go
if value <= g.threshold { return 0.0 }         // Budget() — hard zero once the gate shuts

// pkg/redis/sortedset_impl.go
batchSize := int(math.Floor(float64(r.batchSize) * budget))   // 0 * anything = 0
for i := 0; i < batchSize; i++ {                              // body never runs
    verdict, err := gate.Apply(...)                           // the only Refuse() site
    ...RecordGateDecision(reason, ...)                        // unreachable
}

Back-pressure is applied pre-dequeue by shrinking the batch; the counter was incremented post-dequeue by refusing an already-popped message. The Pub/Sub path has the same shape — currBatchSize <= 0 parks the loop before Receive starts, so the message callback (and its gate.Apply) never runs.

This is worse than a metric that reads zero: a CounterVec label set that has never been incremented is absent from /metrics entirely, so the query returns an empty vector rather than 0. The one metric whose help text promises to answer "is the gate shedding work, and why?" was silent exactly when the gate was doing its job, and operators had to know that async_dispatch_budget == 0 is the real sentinel.

Change

Record the decision where it is actually made.

• Sorted set (processMessages): when the budget shrinks the batch to zero, count one gate_closed decision per throttled poll — but only when ZCARD > 0, since an idle queue was not held back. ZCARD is only issued on throttled polls, so there is no cost on the normal path.
• Pub/Sub (requestWorker): count each receive window skipped for currBatchSize <= 0. There is no cheap depth probe there — the subscription backlog comes from Cloud Monitoring — so this counts the window regardless of what happens to be waiting. The loop re-evaluates every 10s while shut, so the rate is low.
• Pre-create the series. metrics.InitGateDecisions creates all four reason series at 0 when a queue starts, on both backends, so a query for a reason that has not fired yet returns 0 instead of an empty vector.
• Say what the counter means. Its unit is decisions, not messages: quota_exhausted/dropped/error count individual messages refused after dequeue, and gate_closed covers those plus every dequeue round the budget emptied. Updated in both the metric help text and the README metric table.

Also folded the duplicated configMap pool lookup into poolNameFor, now that two call sites need it.

Effect

For the reported run — gate shut for 18% of samples, queue held at a 185 backlog — rate(llm_d_async_async_gate_decisions_total{reason="gate_closed"}[5m]) is now non-zero throughout, at the poll cadence (500ms by default), giving finer resolution of throttling duration than a 15–30s scrape of async_dispatch_budget can. async_dispatch_budget == 0 remains valid and is still the "right now" view; it is no longer the only signal.

Tests

• pkg/redis: TestSortedSetFlow_ClosedGateRecordsGateClosedDecision — a closed gate over an empty queue counts nothing; over a backlogged queue each throttled poll counts one gate_closed, no message is dispatched, and the message stays in Redis.
• pkg/metrics: TestInitGateDecisions — all four reason series exist and read 0 after init.

Compatibility

No config or API surface changes. Existing dashboards and alerts on async_gate_decisions_total keep working; series that were previously absent now appear at 0, and gate_closed starts moving under budget throttling. Anything that treated an absent gate_closed series as "no throttling" was already wrong, and now gets the real answer.

[jtechapps]

I am wondering if we should use the gate owner introduced here #372.

Also I realized we are only recording this metric for queue level gates.

[shimib]

1. Gate Owner: Since the dequeue loops (processMessages and requestWorker) are the original sources of truth for these queue/pool labels, I think it's cleaner to use local variables directly rather than polluting the pipeline.Gate interface with an Owner() getter or using type-casting.
2. Queue-level Only: Yes, pool-level gates are evaluated post-dequeue in the worker thread where individual queue context is lost. Recording pool-level gate decisions (e.g., omitting queue labels) is a great candidate for post-v0.9 work, but out of scope for this bug fix.