feat: expose capacity metrics

README.md

@@ -57,6 +57,10 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费
   `proxy_pool_controller_extraction_returned_proxies_total`；`result` 仅有完成、
   部分、空、库存不足、幂等冲突、限流、不可用、无效与内部错误等固定枚举，
   不包含 Client、请求、过滤条件、Upstream、Proxy 或错误文本标签。
+- **容量指标**：Controller 在既有 Provider 库存对账周期聚合托管 Proxy、可用/有效
+  Slot、待拉取数量和活跃上游数，并暴露
+  `proxy_pool_controller_capacity_inventory_reads_total{result}`；`result` 仅有
+  `success` 和 `error`，不按 Upstream、Worker、Proxy、会话或地址拆分。
 - **PostgreSQL 管理面**：持久化配置版本、Upstream/Routing 管理状态、Admin
   审计与 Outbox；不保存 Proxy 明细或逐次提取记录。
 - **Gateway 组件**：HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、重试、超时、

docs/configuration/reference.md

@@ -526,8 +526,10 @@ Metrics 启用时 `listen` 必须是合法 `host:port`。该入口固定提供 `
 `result` 固定为 accepted、rejected。Gateway 另暴露
 `proxy_pool_gateway_outcomes_total{stage,result}` 与
 `proxy_pool_gateway_outcome_queue_dropped_total`；`stage` 固定为 DIAL、PROXY_HANDSHAKE、
-RESPONSE_HEADERS、TUNNEL，`result` 固定为 success、failure。绝不包含 Proxy、IP、Checker、
+RESPONSE_HEADERS、TUNNEL，`result` 固定为 success、failure。Provider、Extraction、
-Client、路由、目标 URL 或凭据标签。Provider、提取和容量等业务指标仍在后续实施范围。Distribution 启用时 `/readyz` 只以 Redis 活动池为
+Capacity 和 Drain 业务指标也已接入，所有标签均为固定枚举；绝不包含 Proxy、IP、Checker、
+Client、路由、目标 URL 或凭据标签。Capacity 从既有 Provider 库存对账循环聚合，
+不进入 Gateway 热路径。Distribution 启用时 `/readyz` 只以 Redis 活动池为
 服务流量门槛，PostgreSQL 故障由 Admin 接口独立报告。Metrics 开关或监听地址
 变更需要重启 Controller。
 

docs/development/implementation-plan.md

@@ -206,7 +206,7 @@ Distribution/Admin 服务构造、错误合并和资源关闭。生产 Provider
 组合 fixture 已验证隔离 Redis namespace 下的选主、Provider HTTP 调用、模板解析
 和活动池写入。Controller Metrics 独立入口现已提供 `/livez`、`/readyz` 与基础
 Prometheus 运行时指标，三监听器隔离已通过测试；Checker、Gateway、Drain、
-Provider 与 Extraction 业务指标已接入，容量指标和密钥安全的结构化日志仍待实现。双存储 bootstrap 已通过 PostgreSQL 18 + Redis 8.2 组合 fixture，覆盖
+Provider、Extraction 与容量业务指标已接入，密钥安全的结构化日志仍待实现。双存储 bootstrap 已通过 PostgreSQL 18 + Redis 8.2 组合 fixture，覆盖
 迁移、启动配置提交、Readiness、Admin Status 和 Metrics 探针。
 
 WorkerControlPlane 现已接入 Controller 生命周期：Register、ACK 和 Runtime
@@ -355,7 +355,8 @@ Supervisor 也改为同时服从静态配置与管理态，消除两条启停消
 `proxy_pool_controller_provider_valid_candidates_total` 与
 `proxy_pool_controller_provider_new_proxies_total`；`class` 仅允许 `valid`、`empty`、
 `duplicate_only`、`error`。Extraction 还暴露固定 `result` 的请求次数、请求数与响应
-交付数，幂等重放按响应交付统计。容量指标与密钥安全的结构化日志仍待补齐。
+交付数，幂等重放按响应交付统计。容量指标由既有 Provider 库存对账周期聚合，不进入
+Gateway 热路径；仅密钥安全的结构化日志仍待补齐。
 
 ## Task 14: Documentation, Examples, and Diagrams
 

docs/requirements/completion-audit.md

@@ -104,8 +104,8 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。
    响应校验场景；`proxy-checker` 的 BASIC 任务进程已经完成，
    `proxy-controller` 已完成
    Admin/Distribution/Metrics 与 PostgreSQL/Redis 启动装配，`proxy-gateway` 已完成
-   HTTP/Metrics 与控制面 Session 装配；Checker 与 Gateway Outcome 的低基数业务指标
+   HTTP/Metrics 与控制面 Session 装配；Checker、Gateway Outcome、Drain、Provider、
-   已接入，但 Provider、提取和容量指标链未闭环。
+   Extraction 与 Capacity 的低基数业务指标已闭环，密钥安全的结构化日志仍待补齐。
 2. Gateway 的生产连接池调优与代表性流量压测。
 3. Provider 分布式 singleflight/Leader、长期凭据回收和累计额度执行器。
 4. Controller 的 PostgreSQL 连接池、迁移和 pgx Adapter 启动装配已完成；

docs/requirements/traceability.md

@@ -87,5 +87,5 @@
 | OPS-001 | 配置校验后构建不可变快照并原子替换 | 8959-8999 | 100k 索引、版本/epoch 与并发 Apply/Acquire 测试 |
 | OPS-002 | 优雅停机停止新请求/Fetch，等待现有流量后超时关闭 | 8981-9000 | Provider Run 收敛与 `Handler.Shutdown` HTTP 排空、Hijacked CONNECT 超时关闭测试 |
 | OPS-003 | PostgreSQL 只保存管理修订、Upstream/Routing 状态、Admin 审计与 Outbox | 当前会话 | ADR-006、`adminstate` 公用契约和六表 Schema 边界测试；真实 PostgreSQL 契约待完成 |
-| OBS-001 | 指标禁止 Proxy IP、session、Client、完整 URL 高基数标签 | 9001-9029 | Controller Prometheus/探针模块已实现；Checker Collector 暴露固定等级和结果，Gateway Collector 暴露 `proxy_pool_gateway_outcomes_total{stage,result}` 与 `proxy_pool_gateway_outcome_queue_dropped_total`，Drain Collector 暴露 `proxy_pool_controller_drain_candidates_total{reason}` 与 `proxy_pool_controller_drains_started_total{reason}`，其中 reason 固定为 unhealthy/upstream_disabled；Provider Collector 暴露固定 class 的拉取次数与候选计数；Extraction Collector 暴露 `proxy_pool_controller_extraction_requests_total{result}`、请求数与交付数，result 固定为 complete/partial/empty/insufficient/idempotency_conflict/rate_limited/unavailable/invalid/error。注册表测试锁定标签集；容量指标和密钥安全的结构化日志仍待实现 |
+| OBS-001 | 指标禁止 Proxy IP、session、Client、完整 URL 高基数标签 | 9001-9029 | Controller Prometheus/探针模块已实现；Checker Collector 暴露固定等级和结果，Gateway Collector 暴露 `proxy_pool_gateway_outcomes_total{stage,result}` 与 `proxy_pool_gateway_outcome_queue_dropped_total`，Drain Collector 暴露 `proxy_pool_controller_drain_candidates_total{reason}` 与 `proxy_pool_controller_drains_started_total{reason}`，其中 reason 固定为 unhealthy/upstream_disabled；Provider Collector 暴露固定 class 的拉取次数与候选计数；Extraction Collector 暴露固定 result 的请求、请求数与交付数；Capacity Collector 按 Controller 进程聚合库存 Gauge，并只暴露固定 success/error 的库存读取结果，不把 Upstream 写成标签。注册表测试锁定标签集；密钥安全的结构化日志仍待实现 |
 | TEST-001 | 覆盖对话中列出的 11 个关键并发与故障场景 | 9030-9082 | 测试清单；Redis 活动池由 Memory/Redis 公用契约覆盖，跨进程故障场景仍按清单推进 |

findings.md

@@ -54,12 +54,17 @@ Extract 请求使用哪些 Upstream。
 - Provider 拉取结果只按固定 `class` 聚合：`valid`、`empty`、`duplicate_only`、`error`。
   指标分别记录拉取结果次数、合法候选数和新入池代理数，不按 Upstream、Proxy、IP、错误
   文本或凭据拆分。
-- Provider 指标已完成；提取、容量指标和密钥安全的结构化日志仍待实现。
+- Provider 指标已完成；密钥安全的结构化日志仍待实现。
 
 - Extraction 只按固定 `result` 聚合请求：`complete`、`partial`、`empty`、
   `insufficient`、`idempotency_conflict`、`rate_limited`、`unavailable`、`invalid`、
   `error`。计数只记录请求数量和响应交付数量；幂等重放不被误记为新的 Proxy 消费。
-- Extraction 指标已完成；容量指标和密钥安全的结构化日志仍待实现。
+- Extraction 指标已完成；密钥安全的结构化日志仍待实现。
+
+- 容量指标由 Provider 库存对账循环采样，聚合 managed、available、effective、pending
+  与 active Upstream Gauge。Upstream ID 仅保留在 Controller 进程内用于聚合替换和
+  任期结束清理，不输出为 Prometheus 标签；读取结果只允许 `success` 或 `error`。
+- 容量指标已完成；密钥安全的结构化日志仍待实现。
 
 ## 集群与性能
 

internal/controller/bootstrap/bootstrap.go

@@ -77,6 +77,7 @@ type ports struct {
 	coordinator      provider.Coordinator
 	credentials      credentials.Store
 	providerResults  provider.ResultRecorder
+	capacityMetrics  pool.CapacityObserver
 	workerStore      workerruntime.ControlStore
 	close            func() error
 }
@@ -203,6 +204,11 @@ func runWithWorkerFactory(
 		if registrar, ok := opened.providerResults.(provider.ResultObserverRegistrar); ok && !nilInterface(registrar) {
 			registrar.AddResultObserver(providerCollector)
 		}
+		capacityCollector, capacityCollectorErr := platformMetrics.NewCapacityCollector(prometheus.DefaultRegisterer)
+		if capacityCollectorErr != nil {
+			return fmt.Errorf("%w: build capacity metrics: %w", ErrStartup, capacityCollectorErr)
+		}
+		opened.capacityMetrics = capacityCollector
 		extractionCollector, extractionCollectorErr := platformMetrics.NewExtractionCollector(prometheus.DefaultRegisterer)
 		if extractionCollectorErr != nil {
 			return fmt.Errorf("%w: build extraction metrics: %w", ErrStartup, extractionCollectorErr)

internal/controller/bootstrap/provider.go

@@ -75,6 +75,7 @@ func providerRuntimeBuilder(opened ports) (buildUpstreamRuntime, error) {
 			Parser:      parser,
 			Activity:    opened.activity,
 			Results:     results,
+			Capacity:    opened.capacityMetrics,
 		})
 	}, nil
 }

internal/controller/pool/capacity_metrics.go

@@ -0,0 +1,29 @@
+package pool
+
+// CapacityReadResult is the fixed outcome of one authoritative inventory read.
+type CapacityReadResult string
+
+const (
+	CapacityReadSuccess CapacityReadResult = "success"
+	CapacityReadError   CapacityReadResult = "error"
+)
+
+// CapacityObservation contains one Controller-side inventory sample. Its
+// UpstreamID and SourceID are for process-local aggregation only and must not
+// be exposed as Prometheus labels.
+type CapacityObservation struct {
+	UpstreamID      string
+	SourceID        string
+	Result          CapacityReadResult
+	Managed         int
+	AvailableSlots  int64
+	EffectiveSlots  int64
+	PendingExpected int
+}
+
+// CapacityObserver receives cold-path Provider inventory samples. It must
+// return promptly and must not issue another inventory read or persistent I/O.
+type CapacityObserver interface {
+	ObserveCapacity(CapacityObservation)
+	RemoveCapacityUpstream(upstreamID, sourceID string)
+}

internal/controller/provider/upstream_runtime.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"hash/fnv"
 	"reflect"
+	"sync/atomic"
 	"time"
 
 	"proxy-pool/internal/controller/pool"
@@ -30,6 +31,7 @@ type UpstreamRuntimeDependencies struct {
 	Parser      Parser
 	Activity    activitypool.Upserter
 	Results     ResultRecorder
+	Capacity    pool.CapacityObserver
 }
 
 // UpstreamRuntime owns every leader-scoped object for one Upstream. A fresh
@@ -38,6 +40,7 @@ type UpstreamRuntime struct {
 	config       UpstreamRuntimeConfig
 	dependencies UpstreamRuntimeDependencies
 	sleeper      Sleeper
+	termSequence atomic.Uint64
 }
 
 func NewUpstreamRuntime(
@@ -126,26 +129,33 @@ func (runtime *UpstreamRuntime) newLeaderTerm() (*upstreamLeaderTerm, error) {
 	}
 	return &upstreamLeaderTerm{
 		upstreamID:   runtime.config.Provider.UpstreamID,
+		sourceID:     fmt.Sprintf("%p:%d", runtime, runtime.termSequence.Add(1)),
 		interval:     runtime.config.ReconcileInterval,
 		safetyMargin: runtime.config.ReconcilePolicy.SafetyMargin,
 		inventory:    runtime.dependencies.Inventory,
 		pool:         poolReconciler,
 		provider:     providerReconciler,
+		capacity:     runtime.dependencies.Capacity,
 		sleeper:      runtime.sleeper,
 	}, nil
 }
 
 type upstreamLeaderTerm struct {
 	upstreamID   string
+	sourceID     string
 	interval     time.Duration
 	safetyMargin time.Duration
 	inventory    pool.InventoryReader
 	pool         *pool.Reconciler
 	provider     *Reconciler
+	capacity     pool.CapacityObserver
 	sleeper      Sleeper
 }
 
 func (term *upstreamLeaderTerm) run(ctx context.Context, session LeaderSession) error {
+	if !nilRuntimeDependency(term.capacity) {
+		defer term.capacity.RemoveCapacityUpstream(term.upstreamID, term.sourceID)
+	}
 	termCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	done := make(chan error, 2)
@@ -183,7 +193,16 @@ func (term *upstreamLeaderTerm) reconcileInventory(ctx context.Context) error {
 			term.safetyMargin,
 		)
 		if err == nil && inventory.Managed >= 0 && inventory.AvailableSlots >= 0 {
-			term.pool.ReconcileSnapshot(inventory)
+			decision := term.pool.ReconcileSnapshot(inventory)
+			term.observeCapacity(pool.CapacityObservation{
+				UpstreamID: term.upstreamID, SourceID: term.sourceID, Result: pool.CapacityReadSuccess,
+				Managed: inventory.Managed, AvailableSlots: inventory.AvailableSlots,
+				EffectiveSlots: decision.EffectiveSlots, PendingExpected: decision.PendingExpected,
+			})
+		} else {
+			term.observeCapacity(pool.CapacityObservation{
+				UpstreamID: term.upstreamID, SourceID: term.sourceID, Result: pool.CapacityReadError,
+			})
 		}
 		if err := term.sleeper.Sleep(ctx, term.interval); err != nil {
 			if ctx.Err() != nil {
@@ -195,6 +214,13 @@ func (term *upstreamLeaderTerm) reconcileInventory(ctx context.Context) error {
 	return nil
 }
 
+func (term *upstreamLeaderTerm) observeCapacity(observation pool.CapacityObservation) {
+	if term == nil || nilRuntimeDependency(term.capacity) {
+		return
+	}
+	term.capacity.ObserveCapacity(observation)
+}
+
 func initialReconcileDelay(upstreamID string, interval time.Duration) time.Duration {
 	window := min(interval/4, 250*time.Millisecond)
 	if window <= 1 {

internal/controller/provider/upstream_runtime_test.go

@@ -126,6 +126,112 @@ func TestUpstreamRuntimeFailsClosedUntilInventoryReadRecovers(t *testing.T) {
 	}
 }
 
+func TestUpstreamRuntimeReportsCapacityReadsAndClearsTermState(t *testing.T) {
+	observer := newCapacityObserverStub()
+	runtime, err := NewUpstreamRuntime(UpstreamRuntimeConfig{
+		Provider:          runtimeProviderConfig("provider-a"),
+		ReconcilePolicy:   runtimeReconcilePolicy(),
+		ReconcileInterval: 10 * time.Millisecond,
+	}, UpstreamRuntimeDependencies{
+		Coordinator: coordinatorFunc(func(
+			ctx context.Context,
+			_ string,
+			_ CoordinationLimits,
+			work func(context.Context, LeaderSession) error,
+		) error {
+			return work(ctx, unlimitedLeaderSession{})
+		}),
+		Inventory: inventoryReaderFunc(func(context.Context, string, time.Duration) (pool.InventorySnapshot, error) {
+			return pool.InventorySnapshot{Managed: 3, AvailableSlots: 7}, nil
+		}),
+		Adapter: adapterFunc(func(context.Context) (FetchResponse, error) { return FetchResponse{}, nil }),
+		Parser:  parserFunc(func(context.Context, []byte) ([]proxyDomain.Proxy, error) { return nil, nil }),
+		Activity: activitySinkFunc(func(context.Context, string, activitypool.FetchedBatch) (activitypool.UpsertResult, error) {
+			return activitypool.UpsertResult{}, nil
+		}),
+		Results:  resultRecorderFunc(func(Result) {}),
+		Capacity: observer,
+	})
+	if err != nil {
+		t.Fatalf("NewUpstreamRuntime() error = %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- runtime.Run(ctx) }()
+	select {
+	case observed := <-observer.observations:
+		if observed.UpstreamID != "provider-a" || observed.SourceID == "" || observed.Result != pool.CapacityReadSuccess ||
+			observed.Managed != 3 || observed.AvailableSlots != 7 || observed.EffectiveSlots != 7 || observed.PendingExpected != 0 {
+			t.Fatalf("capacity observation = %+v", observed)
+		}
+	case <-time.After(time.Second):
+		cancel()
+		<-done
+		t.Fatal("timed out waiting for capacity observation")
+	}
+	cancel()
+	if err := <-done; !errors.Is(err, context.Canceled) {
+		t.Fatalf("Run() error = %v, want context cancellation", err)
+	}
+	select {
+	case removal := <-observer.removed:
+		if removal.upstreamID != "provider-a" || removal.sourceID == "" {
+			t.Fatalf("capacity removal = %+v", removal)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for capacity cleanup")
+	}
+}
+
+func TestUpstreamRuntimeReportsFailedCapacityReads(t *testing.T) {
+	observer := newCapacityObserverStub()
+	runtime, err := NewUpstreamRuntime(UpstreamRuntimeConfig{
+		Provider:          runtimeProviderConfig("provider-a"),
+		ReconcilePolicy:   runtimeReconcilePolicy(),
+		ReconcileInterval: 10 * time.Millisecond,
+	}, UpstreamRuntimeDependencies{
+		Coordinator: coordinatorFunc(func(
+			ctx context.Context,
+			_ string,
+			_ CoordinationLimits,
+			work func(context.Context, LeaderSession) error,
+		) error {
+			return work(ctx, unlimitedLeaderSession{})
+		}),
+		Inventory: inventoryReaderFunc(func(context.Context, string, time.Duration) (pool.InventorySnapshot, error) {
+			return pool.InventorySnapshot{}, errors.New("redis unavailable")
+		}),
+		Adapter: adapterFunc(func(context.Context) (FetchResponse, error) { return FetchResponse{}, nil }),
+		Parser:  parserFunc(func(context.Context, []byte) ([]proxyDomain.Proxy, error) { return nil, nil }),
+		Activity: activitySinkFunc(func(context.Context, string, activitypool.FetchedBatch) (activitypool.UpsertResult, error) {
+			return activitypool.UpsertResult{}, nil
+		}),
+		Results:  resultRecorderFunc(func(Result) {}),
+		Capacity: observer,
+	})
+	if err != nil {
+		t.Fatalf("NewUpstreamRuntime() error = %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- runtime.Run(ctx) }()
+	select {
+	case observed := <-observer.observations:
+		if observed.UpstreamID != "provider-a" || observed.SourceID == "" || observed.Result != pool.CapacityReadError ||
+			observed.Managed != 0 || observed.AvailableSlots != 0 || observed.EffectiveSlots != 0 || observed.PendingExpected != 0 {
+			t.Fatalf("capacity observation = %+v", observed)
+		}
+	case <-time.After(time.Second):
+		cancel()
+		<-done
+		t.Fatal("timed out waiting for failed capacity observation")
+	}
+	cancel()
+	<-done
+}
+
 func TestUpstreamRuntimeDelegatesRequestIntervalOnlyToCoordinator(t *testing.T) {
 	providerConfig := runtimeProviderConfig("provider-a")
 	providerConfig.RequestInterval = 500 * time.Millisecond
@@ -245,3 +351,28 @@ func (f inventoryReaderFunc) ReadInventory(
 ) (pool.InventorySnapshot, error) {
 	return f(ctx, upstreamID, safetyMargin)
 }
+
+type capacityObserverStub struct {
+	observations chan pool.CapacityObservation
+	removed      chan capacityRemoval
+}
+
+type capacityRemoval struct {
+	upstreamID string
+	sourceID   string
+}
+
+func newCapacityObserverStub() *capacityObserverStub {
+	return &capacityObserverStub{
+		observations: make(chan pool.CapacityObservation, 4),
+		removed:      make(chan capacityRemoval, 2),
+	}
+}
+
+func (observer *capacityObserverStub) ObserveCapacity(observation pool.CapacityObservation) {
+	observer.observations <- observation
+}
+
+func (observer *capacityObserverStub) RemoveCapacityUpstream(upstreamID, sourceID string) {
+	observer.removed <- capacityRemoval{upstreamID: upstreamID, sourceID: sourceID}
+}

internal/platform/metrics/capacity.go

@@ -0,0 +1,160 @@
+package metrics
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+
+	controllerPool "proxy-pool/internal/controller/pool"
+)
+
+// CapacityCollector aggregates Controller leader-term inventory samples. An
+// upstream ID is retained only in process memory for replacement and removal;
+// it is never emitted as a Prometheus label.
+type CapacityCollector struct {
+	mu         sync.Mutex
+	byUpstream map[string]capacityState
+
+	reads           *prometheus.CounterVec
+	managed         *prometheus.Desc
+	available       *prometheus.Desc
+	effective       *prometheus.Desc
+	pending         *prometheus.Desc
+	activeUpstreams *prometheus.Desc
+}
+
+type capacityState struct {
+	sourceID  string
+	managed   int
+	available int64
+	effective int64
+	pending   int
+}
+
+var (
+	_ controllerPool.CapacityObserver = (*CapacityCollector)(nil)
+	_ prometheus.Collector            = (*CapacityCollector)(nil)
+)
+
+func NewCapacityCollector(registerer prometheus.Registerer) (*CapacityCollector, error) {
+	if registerer == nil {
+		return nil, ErrInvalidDependencies
+	}
+	reads, err := registerCounterVec(registerer, prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "proxy_pool", Subsystem: "controller", Name: "capacity_inventory_reads_total",
+		Help: "Number of Provider inventory reads by fixed result.",
+	}, []string{"result"}))
+	if err != nil {
+		return nil, err
+	}
+	candidate := &CapacityCollector{
+		byUpstream: map[string]capacityState{}, reads: reads,
+		managed: prometheus.NewDesc(
+			"proxy_pool_controller_capacity_managed_proxies",
+			"Managed non-extracted proxies from the latest successful samples of active Controller Provider terms.", nil, nil,
+		),
+		available: prometheus.NewDesc(
+			"proxy_pool_controller_capacity_available_slots",
+			"Available proxy concurrency slots from the latest successful samples of active Controller Provider terms.", nil, nil,
+		),
+		effective: prometheus.NewDesc(
+			"proxy_pool_controller_capacity_effective_slots",
+			"Available slots plus expected pending fetch capacity from the latest successful active Controller samples.", nil, nil,
+		),
+		pending: prometheus.NewDesc(
+			"proxy_pool_controller_capacity_pending_expected_proxies",
+			"Expected proxy candidates reserved by in-flight fetches in the latest successful active Controller samples.", nil, nil,
+		),
+		activeUpstreams: prometheus.NewDesc(
+			"proxy_pool_controller_capacity_active_upstreams",
+			"Number of Upstreams with a successful inventory sample in this Controller process.", nil, nil,
+		),
+	}
+	return registerCapacityCollector(registerer, candidate)
+}
+
+func (collector *CapacityCollector) ObserveCapacity(observation controllerPool.CapacityObservation) {
+	if collector == nil || collector.reads == nil || observation.UpstreamID == "" || observation.SourceID == "" {
+		return
+	}
+	switch observation.Result {
+	case controllerPool.CapacityReadError:
+		collector.reads.WithLabelValues(string(observation.Result)).Inc()
+		return
+	case controllerPool.CapacityReadSuccess:
+		if observation.Managed < 0 || observation.AvailableSlots < 0 || observation.EffectiveSlots < 0 || observation.PendingExpected < 0 {
+			return
+		}
+		collector.reads.WithLabelValues(string(observation.Result)).Inc()
+	default:
+		return
+	}
+	collector.mu.Lock()
+	collector.byUpstream[observation.UpstreamID] = capacityState{
+		sourceID: observation.SourceID,
+		managed:  observation.Managed, available: observation.AvailableSlots,
+		effective: observation.EffectiveSlots, pending: observation.PendingExpected,
+	}
+	collector.mu.Unlock()
+}
+
+func (collector *CapacityCollector) RemoveCapacityUpstream(upstreamID, sourceID string) {
+	if collector == nil || upstreamID == "" || sourceID == "" {
+		return
+	}
+	collector.mu.Lock()
+	if state, exists := collector.byUpstream[upstreamID]; exists && state.sourceID == sourceID {
+		delete(collector.byUpstream, upstreamID)
+	}
+	collector.mu.Unlock()
+}
+
+func (collector *CapacityCollector) Describe(descriptions chan<- *prometheus.Desc) {
+	if collector == nil {
+		return
+	}
+	descriptions <- collector.managed
+	descriptions <- collector.available
+	descriptions <- collector.effective
+	descriptions <- collector.pending
+	descriptions <- collector.activeUpstreams
+}
+
+func (collector *CapacityCollector) Collect(metrics chan<- prometheus.Metric) {
+	if collector == nil {
+		return
+	}
+	collector.mu.Lock()
+	var managed, available, effective, pending int64
+	for _, state := range collector.byUpstream {
+		managed += int64(state.managed)
+		available += state.available
+		effective += state.effective
+		pending += int64(state.pending)
+	}
+	activeUpstreams := len(collector.byUpstream)
+	collector.mu.Unlock()
+	metrics <- prometheus.MustNewConstMetric(collector.managed, prometheus.GaugeValue, float64(managed))
+	metrics <- prometheus.MustNewConstMetric(collector.available, prometheus.GaugeValue, float64(available))
+	metrics <- prometheus.MustNewConstMetric(collector.effective, prometheus.GaugeValue, float64(effective))
+	metrics <- prometheus.MustNewConstMetric(collector.pending, prometheus.GaugeValue, float64(pending))
+	metrics <- prometheus.MustNewConstMetric(collector.activeUpstreams, prometheus.GaugeValue, float64(activeUpstreams))
+}
+
+func registerCapacityCollector(registerer prometheus.Registerer, candidate *CapacityCollector) (*CapacityCollector, error) {
+	if err := registerer.Register(candidate); err == nil {
+		return candidate, nil
+	} else {
+		var registered prometheus.AlreadyRegisteredError
+		if !errors.As(err, &registered) {
+			return nil, fmt.Errorf("register capacity collector: %w", err)
+		}
+		existing, ok := registered.ExistingCollector.(*CapacityCollector)
+		if !ok {
+			return nil, fmt.Errorf("register capacity collector: existing collector has unexpected type")
+		}
+		return existing, nil
+	}
+}

internal/platform/metrics/capacity_test.go

@@ -0,0 +1,115 @@
+package metrics
+
+import (
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+
+	controllerPool "proxy-pool/internal/controller/pool"
+)
+
+func TestCapacityCollectorAggregatesActiveUpstreamsWithoutLabels(t *testing.T) {
+	registry := prometheus.NewRegistry()
+	collector, err := NewCapacityCollector(registry)
+	if err != nil {
+		t.Fatalf("NewCapacityCollector() error = %v", err)
+	}
+	collector.ObserveCapacity(controllerPool.CapacityObservation{
+		UpstreamID: "provider-a", SourceID: "term-a", Result: controllerPool.CapacityReadSuccess,
+		Managed: 2, AvailableSlots: 5, EffectiveSlots: 7, PendingExpected: 2,
+	})
+	collector.ObserveCapacity(controllerPool.CapacityObservation{
+		UpstreamID: "provider-b", SourceID: "term-b", Result: controllerPool.CapacityReadSuccess,
+		Managed: 3, AvailableSlots: 4, EffectiveSlots: 6, PendingExpected: 1,
+	})
+	collector.ObserveCapacity(controllerPool.CapacityObservation{
+		UpstreamID: "provider-a", SourceID: "term-a", Result: controllerPool.CapacityReadError,
+	})
+	collector.ObserveCapacity(controllerPool.CapacityObservation{
+		UpstreamID: "client-a", SourceID: "term-client", Result: "invalid", Managed: 100,
+	})
+
+	assertMetricValue(t, registry, "proxy_pool_controller_capacity_inventory_reads_total", map[string]string{"result": "success"}, 2)
+	assertMetricValue(t, registry, "proxy_pool_controller_capacity_inventory_reads_total", map[string]string{"result": "error"}, 1)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_managed_proxies", 5)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_available_slots", 9)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_effective_slots", 13)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_pending_expected_proxies", 3)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_active_upstreams", 2)
+
+	collector.RemoveCapacityUpstream("provider-a", "term-a")
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_managed_proxies", 3)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_available_slots", 4)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_effective_slots", 6)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_pending_expected_proxies", 1)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_active_upstreams", 1)
+}
+
+func TestNewCapacityCollectorReusesRegisteredCollectors(t *testing.T) {
+	registry := prometheus.NewRegistry()
+	first, err := NewCapacityCollector(registry)
+	if err != nil {
+		t.Fatalf("first NewCapacityCollector() error = %v", err)
+	}
+	second, err := NewCapacityCollector(registry)
+	if err != nil {
+		t.Fatalf("second NewCapacityCollector() error = %v", err)
+	}
+	first.ObserveCapacity(controllerPool.CapacityObservation{
+		UpstreamID: "provider-a", SourceID: "term-a", Result: controllerPool.CapacityReadSuccess,
+		Managed: 1, AvailableSlots: 2, EffectiveSlots: 3,
+	})
+	second.ObserveCapacity(controllerPool.CapacityObservation{
+		UpstreamID: "provider-b", SourceID: "term-b", Result: controllerPool.CapacityReadSuccess,
+		Managed: 2, AvailableSlots: 3, EffectiveSlots: 5, PendingExpected: 1,
+	})
+
+	assertMetricValue(t, registry, "proxy_pool_controller_capacity_inventory_reads_total", map[string]string{"result": "success"}, 2)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_managed_proxies", 3)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_available_slots", 5)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_effective_slots", 8)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_pending_expected_proxies", 1)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_active_upstreams", 2)
+}
+
+func TestCapacityCollectorKeepsNewTermWhenOldTermStops(t *testing.T) {
+	registry := prometheus.NewRegistry()
+	collector, err := NewCapacityCollector(registry)
+	if err != nil {
+		t.Fatalf("NewCapacityCollector() error = %v", err)
+	}
+	collector.ObserveCapacity(controllerPool.CapacityObservation{
+		UpstreamID: "provider-a", SourceID: "old-term", Result: controllerPool.CapacityReadSuccess,
+		Managed: 1, AvailableSlots: 2, EffectiveSlots: 3,
+	})
+	collector.ObserveCapacity(controllerPool.CapacityObservation{
+		UpstreamID: "provider-a", SourceID: "new-term", Result: controllerPool.CapacityReadSuccess,
+		Managed: 4, AvailableSlots: 5, EffectiveSlots: 6, PendingExpected: 1,
+	})
+	collector.RemoveCapacityUpstream("provider-a", "old-term")
+
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_managed_proxies", 4)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_available_slots", 5)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_effective_slots", 6)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_pending_expected_proxies", 1)
+	assertGaugeValue(t, registry, "proxy_pool_controller_capacity_active_upstreams", 1)
+}
+
+func assertGaugeValue(t *testing.T, registry *prometheus.Registry, name string, want float64) {
+	t.Helper()
+	metrics, err := registry.Gather()
+	if err != nil {
+		t.Fatalf("Gather() = %v", err)
+	}
+	for _, family := range metrics {
+		if family.GetName() != name {
+			continue
+		}
+		for _, metric := range family.GetMetric() {
+			if len(metric.GetLabel()) == 0 && metric.GetGauge().GetValue() == want {
+				return
+			}
+		}
+	}
+	t.Fatalf("gauge %s value=%v was not found", name, want)
+}

progress.md

@@ -2,17 +2,22 @@
 
 ## 2026-08-02
 
+- 容量可观测性已接入 Provider 补池对账循环：每个成功 Redis 库存读数更新 Controller
+  进程级 managed/available/effective/pending/active Gauge，失败只增加固定
+  `success`/`error` 的读取计数。Upstream 仅用于进程内聚合，领导任期结束即清除，
+  不成为 Prometheus 标签；不进入 Gateway 10 万 QPS 容量热路径，也不增加 Redis 查询。
+  密钥安全的结构化日志仍待实现。
 - Extraction 可观测性已接入公用 `MetricsObserver`：Controller 暴露固定 `result`
   的请求计数、请求 Proxy 总数和响应交付 Proxy 总数。事件在服务单一出口分类，覆盖
   complete/partial/empty/insufficient/idempotency_conflict/rate_limited/unavailable/
   invalid/error；不带 Client、Request、Filter、Upstream、Proxy 或错误文本标签。
-  幂等重放统计响应交付而不误计为新的 Redis 消费。容量指标和密钥安全结构化日志仍待实现。
+  幂等重放统计响应交付而不误计为新的 Redis 消费。密钥安全结构化日志仍待实现。
 - Provider 拉取可观测性已接入公用 `ResultObserver`：Controller 暴露
   `proxy_pool_controller_provider_fetch_results_total{class}`、
   `proxy_pool_controller_provider_valid_candidates_total` 与
   `proxy_pool_controller_provider_new_proxies_total`。`class` 仅允许 `valid`、`empty`、
   `duplicate_only`、`error`；指标不包含 Upstream、Proxy、IP、错误文本或凭据标签。
-  Provider 指标已完成，提取、容量指标和密钥安全的结构化日志仍待实现。
+  Provider 指标已完成；密钥安全的结构化日志仍待实现。
 - 已实现管理态驱动的即时完整快照刷新：`SnapshotRefreshBroker` 以每个 Worker 流一格的
   非阻塞通道广播，连续变更自动合并。Admin 成功提交 Upstream 启停、Routing 切换和配置
   发布时触发通知；Worker 保留原有定时刷新作为失效保护。Routing 切换不按旧 Upstream