diff --git a/docs/design/architecture.md b/docs/design/architecture.md index 6df8135..e1459da 100644 --- a/docs/design/architecture.md +++ b/docs/design/architecture.md @@ -150,8 +150,9 @@ Worker 以有界周期批量上报运行态，而不是在每个 Gateway 请求 继续以 draining 状态上报。Controller 使用 session、单调 report sequence、 ownership epoch、Controller 已 ACK 的 snapshot/epoch 上界和 Redis 服务端 TTL 共同校验；缺失、过期或超前报告按零可用容量 fail-closed。运行态报告扫描有界 -当前 Snapshot，并以分片索引补充已移除但仍非零的 runtime；历史 Capacity -注册表有硬上限，避免短 TTL Proxy 持续轮换导致心跳扫描与内存无界增长。 +当前 Snapshot，并以分片索引补充已移除但仍非零的 runtime；移除 Proxy 会先停止新 +预留，Active/Reserved 归零后在后续 Snapshot Apply 回收历史 Capacity。注册表仍有 +硬上限，避免短 TTL Proxy 持续轮换导致心跳扫描与内存无界增长。 ## 6. Proxy 状态机 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index c2349f5..156e9a2 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -85,8 +85,9 @@ test/{fixtures,integration,e2e,load}/ - [ ] Add race coverage and duplicate-release invariant metrics hook. 当前进度（2026-07-29）：固定 Max 下的每 Proxy 打包 CAS、Cancel/Commit/Release -生命周期、重复终结、错误顺序和同一 Reservation 并发终结已通过领域测试；动态 -降容契约、低基数不变量指标、Linux race 证据及短 TTL runtime 排空回收待完成。 +生命周期、重复终结、错误顺序和同一 Reservation 并发终结已通过领域测试；已退出 +Proxy 会停止新预留、保留非零 Drain 计数，并在归零后由后续 Snapshot Apply 回收。 +动态降容契约、低基数不变量指标和 Linux race 证据仍待完成。 ## Task 4: Routing and Sequential Switching diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index f9a6424..9b3b57c 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -106,8 +106,8 @@ CI 已配置 Linux race job。PostgreSQL 18 和 Redis 8.2 的隔离 Adapter fixt 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 10. 将五种 Routing 策略和 `onUnavailable` 接入 Gateway/Distribution 运行链， 补齐 Sequential 持久化恢复、跨实例 CAS 和 disabled candidate 语义。 -11. 补齐 Proxy Capacity 动态降容契约、Reservation 全生命周期观测及短 TTL - Proxy 运行态排空回收。 +11. 补齐 Proxy Capacity 动态降容契约、Reservation 全生命周期观测；短 TTL + Proxy 的零计数运行态已在后续 Snapshot Apply 中回收。 ## 4. 容量结论 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index e519edc..0ba234b 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -45,7 +45,7 @@ | PROXY-001 | Proxy 保存协议、地址、凭据引用、来源、TTL、健康、容量和标签 | 71-105, 8605-8678 | Domain 类型与序列化测试 | | PROXY-002 | 唯一键包含 scheme、host、port、username、credentialVersion | 6655-6727, 8605-8678 | 去重单测 | | PROXY-003 | TTL 来源优先级明确并统一 UTC | 681-747, 8655-8678 | TTL 表驱动测试 | -| CAP-001 | Gateway 分配使用 Reserved -> Active 原子转换 | 1203-1467, 8530-8597 | 固定 Max 下打包 CAS 与 1,000 并发不超卖已完成；动态降容和完整生命周期证据待完成 | +| CAP-001 | Gateway 分配使用 Reserved -> Active 原子转换 | 1203-1467, 8530-8597 | 固定 Max 下打包 CAS、1,000 并发不超卖、退役 Proxy 停止新预留及零计数历史运行态回收已完成；动态降容和完整生命周期证据待完成 | | CAP-002 | 补池依据 Available Slots，不只看 Proxy 数量 | 1203-1402, 8530-8597 | `AvailableSlots`、显式 minimum/target 水位、pending 槽位和迟滞 Reconciler 已测试；Worker session/ACK/sequence/TTL/ownership fence、基础 Snapshot 流及 Gateway 会话组件已通过内存、Redis 8.2 和 gRPC 测试；权威 payload、目标健康和 Gateway reserve 策略仍待完成 | | CAP-003 | pool.maxSize 包括 FETCHED/CHECKING/AVAILABLE/SUSPECT/DRAINING 与 pending expected | 3001-3533, 6642-6680 | `FetchBudget` 100 并发额度预占测试 | | CAP-004 | TTL safety margin 内禁止新分配 | 173-220, 6728-6741 | 时钟测试 | diff --git a/internal/domain/proxy/capacity.go b/internal/domain/proxy/capacity.go index 0369fef..2913027 100644 --- a/internal/domain/proxy/capacity.go +++ b/internal/domain/proxy/capacity.go @@ -15,6 +15,8 @@ var ( type Capacity struct { max atomic.Uint32 counters atomic.Uint64 + reservationEnabled atomic.Bool + reserving atomic.Int64 configuredObserver *activityObserver observer atomic.Pointer[activityObserver] } @@ -33,6 +35,7 @@ func NewCapacityWithActivityObserver(max int64, observer func(nonzero bool)) *Ca max = 0 } capacity.max.Store(uint32(max)) + capacity.reservationEnabled.Store(true) if observer != nil { capacity.configuredObserver = &activityObserver{notify: observer} capacity.observer.Store(capacity.configuredObserver) @@ -53,6 +56,25 @@ func (c *Capacity) SetActivityObservationEnabled(enabled bool) { c.observer.Store(nil) } +// SetReservationEnabled gates new reservations without changing existing +// Active or Reserved counters. It is used while a snapshot retires a Proxy. +func (c *Capacity) SetReservationEnabled(enabled bool) { + if c == nil { + return + } + c.reservationEnabled.Store(enabled) +} + +// Reclaimable reports whether a retired capacity can be forgotten without +// losing an in-flight reservation or draining runtime counter. +func (c *Capacity) Reclaimable() bool { + if c == nil || c.reservationEnabled.Load() || c.reserving.Load() != 0 { + return false + } + active, reserved, _ := c.Counters() + return active == 0 && reserved == 0 +} + func (c *Capacity) SetMax(max int64) bool { if max < 0 || max > int64(counterMask) { return false @@ -73,7 +95,14 @@ func (c *Capacity) Counters() (active, reserved, maximum int64) { } func (c *Capacity) Reserve() (*Reservation, bool) { + if c == nil || !c.beginReservation() { + return nil, false + } + defer c.reserving.Add(-1) for { + if !c.reservationEnabled.Load() { + return nil, false + } current := c.counters.Load() active, reserved := unpack(current) if active+reserved >= c.max.Load() { @@ -81,6 +110,10 @@ func (c *Capacity) Reserve() (*Reservation, bool) { } next := pack(active, reserved+1) if c.counters.CompareAndSwap(current, next) { + if !c.reservationEnabled.Load() { + c.cancel() + return nil, false + } if active+reserved == 0 { if observer := c.observer.Load(); observer != nil { observer.notify(true) @@ -91,6 +124,18 @@ func (c *Capacity) Reserve() (*Reservation, bool) { } } +func (c *Capacity) beginReservation() bool { + if !c.reservationEnabled.Load() { + return false + } + c.reserving.Add(1) + if c.reservationEnabled.Load() { + return true + } + c.reserving.Add(-1) + return false +} + func (c *Capacity) Active() int64 { active, _ := unpack(c.counters.Load()) return int64(active) diff --git a/internal/domain/proxy/capacity_test.go b/internal/domain/proxy/capacity_test.go index fe36d91..f7e3a3a 100644 --- a/internal/domain/proxy/capacity_test.go +++ b/internal/domain/proxy/capacity_test.go @@ -89,6 +89,30 @@ func TestCapacityActivityObserverTracksOnlyNonzeroTransitions(t *testing.T) { } } +func TestCapacityRetirementStopsNewReservationsAndRetainsActiveWork(t *testing.T) { + capacity := NewCapacity(1) + reservation, ok := capacity.Reserve() + if !ok { + t.Fatal("Reserve() = false") + } + if err := reservation.Commit(); err != nil { + t.Fatalf("Commit(): %v", err) + } + capacity.SetReservationEnabled(false) + if _, ok := capacity.Reserve(); ok { + t.Fatal("Reserve() after retirement = true, want false") + } + if capacity.Reclaimable() { + t.Fatal("Reclaimable() with active work = true, want false") + } + if err := reservation.Release(); err != nil { + t.Fatalf("Release(): %v", err) + } + if !capacity.Reclaimable() { + t.Fatal("Reclaimable() after release = false, want true") + } +} + func TestReservationCommitAndReleaseAreSingleUse(t *testing.T) { capacity := NewCapacity(1) reservation, ok := capacity.Reserve() diff --git a/internal/gateway/snapshot/store.go b/internal/gateway/snapshot/store.go index eef21b4..33aed6d 100644 --- a/internal/gateway/snapshot/store.go +++ b/internal/gateway/snapshot/store.go @@ -246,23 +246,41 @@ func (s *Store) Apply(envelope Envelope) error { } proxies := cloneAndSort(envelope.Proxies) + nextProxyIDs := make(map[string]struct{}, len(proxies)) + for _, descriptor := range proxies { + nextProxyIDs[descriptor.ID] = struct{}{} + } + + retiredCurrent := make(map[string]*runtimeRegistration) + if current != nil { + for _, entry := range current.Entries { + if _, retained := nextProxyIDs[entry.Proxy.ID]; retained { + continue + } + registration := s.runtimes[entry.Proxy.ID] + registration.capacity.SetReservationEnabled(false) + registration.current.Store(false) + registration.capacity.SetActivityObservationEnabled(true) + active, reserved, _ := registration.capacity.Counters() + s.active.track(entry.Proxy.ID, registration.capacity, active+reserved > 0) + retiredCurrent[entry.Proxy.ID] = registration + } + } + + reclaimable := s.reclaimableRuntimes(nextProxyIDs) newRuntimeCount := 0 for _, descriptor := range proxies { if s.runtimes[descriptor.ID] == nil { newRuntimeCount++ } } - if len(s.runtimes)+newRuntimeCount > s.limit { + if len(s.runtimes)-len(reclaimable)+newRuntimeCount > s.limit { + s.restoreCurrentRuntimes(retiredCurrent) return ErrRuntimeLimitExceeded } - if current != nil { - for _, entry := range current.Entries { - registration := s.runtimes[entry.Proxy.ID] - registration.capacity.SetActivityObservationEnabled(true) - registration.current.Store(false) - active, reserved, _ := registration.capacity.Counters() - s.active.track(entry.Proxy.ID, registration.capacity, active+reserved > 0) - } + for proxyID, registration := range reclaimable { + s.active.remove(proxyID, registration.capacity) + delete(s.runtimes, proxyID) } entries := make([]Entry, 0, len(proxies)) for _, descriptor := range proxies { @@ -281,10 +299,10 @@ func (s *Store) Apply(envelope Envelope) error { } registration.current.Store(true) registration.capacity.SetActivityObservationEnabled(false) + registration.capacity.SetReservationEnabled(true) s.active.remove(descriptor.ID, registration.capacity) - // Keep runtimes for temporarily absent IDs. Old immutable views may still - // hold in-flight leases, so reclaiming here could reset active capacity if - // the same Proxy reappears in a later snapshot. + // A retired runtime remains until Active/Reserved reaches zero. Once it is + // quiescent, the next Apply reclaims it before enforcing the registry limit. s.runtimes[descriptor.ID] = registration entries = append(entries, Entry{Proxy: descriptor, Runtime: registration.capacity}) } @@ -303,6 +321,26 @@ func (s *Store) Apply(envelope Envelope) error { return nil } +func (s *Store) reclaimableRuntimes(nextProxyIDs map[string]struct{}) map[string]*runtimeRegistration { + result := make(map[string]*runtimeRegistration) + for proxyID, registration := range s.runtimes { + if _, retained := nextProxyIDs[proxyID]; retained || registration.current.Load() || !registration.capacity.Reclaimable() { + continue + } + result[proxyID] = registration + } + return result +} + +func (s *Store) restoreCurrentRuntimes(retired map[string]*runtimeRegistration) { + for proxyID, registration := range retired { + registration.current.Store(true) + registration.capacity.SetActivityObservationEnabled(false) + registration.capacity.SetReservationEnabled(true) + s.active.remove(proxyID, registration.capacity) + } +} + func (v *View) Select(query Query) Selection { if query.Now.IsZero() { query.Now = time.Now().UTC() diff --git a/internal/gateway/snapshot/store_test.go b/internal/gateway/snapshot/store_test.go index bd6766d..7c5b789 100644 --- a/internal/gateway/snapshot/store_test.go +++ b/internal/gateway/snapshot/store_test.go @@ -350,7 +350,7 @@ func TestStoreRuntimeReportMarksCurrentDrainingProxy(t *testing.T) { } } -func TestStoreBoundsHistoricalRuntimeRegistryAndKeepsApplyTransactional(t *testing.T) { +func TestStoreReclaimsRetiredZeroRuntimeBeforeHittingLimit(t *testing.T) { store, err := NewStoreWithRuntimeLimit("cluster-a", "worker-a", 1) if err != nil { t.Fatalf("NewStoreWithRuntimeLimit(): %v", err) @@ -381,11 +381,64 @@ func TestStoreBoundsHistoricalRuntimeRegistryAndKeepsApplyTransactional(t *testi Full: true, Proxies: []proxyDomain.Proxy{second}, } overLimit.Checksum = Checksum(overLimit.Proxies) + if err := store.Apply(overLimit); err != nil { + t.Fatalf("Apply(replacement after retired runtime) error = %v", err) + } + if current := store.Current(); current.Version != 3 || len(current.Entries) != 1 || current.Entries[0].Proxy.ID != "proxy-b" { + t.Fatalf("Current() after replacement = %+v", current) + } + if len(store.runtimes) != 1 || store.runtimes["proxy-b"] == nil || store.runtimes["proxy-a"] != nil { + t.Fatalf("runtime registry = %+v, want only proxy-b", store.runtimes) + } +} + +func TestStoreRetainsActiveRuntimeAndKeepsApplyTransactionalAtLimit(t *testing.T) { + store, err := NewStoreWithRuntimeLimit("cluster-a", "worker-a", 1) + if err != nil { + t.Fatalf("NewStoreWithRuntimeLimit(): %v", err) + } + first := proxyDomain.Proxy{ + ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, + State: proxyDomain.StateAvailable, MaxConcurrency: 2, + } + envelope := Envelope{ + ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 1, + Full: true, Proxies: []proxyDomain.Proxy{first}, + } + envelope.Checksum = Checksum(envelope.Proxies) + if err := store.Apply(envelope); err != nil { + t.Fatalf("Apply(first): %v", err) + } + reservation, ok := store.Current().Entries[0].Runtime.Reserve() + if !ok { + t.Fatal("Reserve() = false") + } + if err := reservation.Commit(); err != nil { + t.Fatalf("Commit(): %v", err) + } + second := first + second.ID = "proxy-b" + overLimit := Envelope{ + ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 2, + Full: true, Proxies: []proxyDomain.Proxy{second}, + } + overLimit.Checksum = Checksum(overLimit.Proxies) if err := store.Apply(overLimit); !errors.Is(err, ErrRuntimeLimitExceeded) { t.Fatalf("Apply(over limit) error = %v, want ErrRuntimeLimitExceeded", err) } - if current := store.Current(); current.Version != 2 || len(current.Entries) != 0 { - t.Fatalf("Current() after rejected apply = version %d entries %d", current.Version, len(current.Entries)) + current := store.Current() + if current.Version != 1 || len(current.Entries) != 1 || current.Entries[0].Proxy.ID != "proxy-a" { + t.Fatalf("Current() after rejected apply = %+v", current) + } + extra, ok := current.Entries[0].Runtime.Reserve() + if !ok { + t.Fatal("Reserve() after rejected Apply = false, want current Proxy to remain enabled") + } + if err := extra.Cancel(); err != nil { + t.Fatalf("Cancel(extra): %v", err) + } + if err := reservation.Release(); err != nil { + t.Fatalf("Release(): %v", err) } }