youfak/proxy-pool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: reap sustained unhealthy proxies
Some checks are pending
ci / proto (push) Waiting to run
Details
ci / test (ubuntu-latest) (push) Waiting to run
Details
ci / test (windows-latest) (push) Waiting to run
Details
ci / race (push) Waiting to run
Details
ci / integration (push) Waiting to run
Details

Browse Source
This commit is contained in:
youfak 2026-08-02 10:15:25 +08:00
parent 2af504fd49
commit 84ed10bd7a
37 changed files with 689 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@ -258,8 +258,8 @@ go run ./cmd/proxy-loadgen `
- **P0 - Worker 控制面闭环**：Worker session、Snapshot ledger、ACK、运行态接收、
ownership 索引，以及 Gateway 快照客户端。
- **P0 - Checker 健康链**：BASIC/EGRESS/TARGET 的共享调度、实际探测、Observation reducer 和
`FETCHED -> AVAILABLE / SUSPECT / UNHEALTHY` 状态链已完成；TARGET 事实按路由目标 Profile
独立归并。
`FETCHED -> AVAILABLE / SUSPECT / UNHEALTHY` 状态链已完成；可配置的持续 UNHEALTHY
回收只删除未分配 Proxy，已分配项等待 Drain/ACK；TARGET 事实按路由目标 Profile 独立归并。
- **P1 - Gateway 与 Routing**：Gateway 进程、快照凭据分发、五种 Routing 策略与
`onUnavailable` 已接入；动态容量调整和 Drain 闭环待完成。
- **P1 - 可观测与部署**：低基数业务指标、完整 Compose/Kubernetes 进程拓扑，

8
docs/configuration/reference.md
View File

@ -372,6 +372,7 @@ upstreams:
timeout: 2s
maxAttempts: 2
maxConsecutiveFailures: 3
unhealthyRemoveAfter: 0s
urls: [http://connect.rom.miui.com/generate_204]
```
@ -477,6 +478,13 @@ proxyAuth:
配置多个目标后产生无界检查流量。
- 第一次有意义失败进入 SUSPECT；达到 `maxConsecutiveFailures` 后才进入
UNHEALTHY。
- `unhealthyRemoveAfter` 控制 UNHEALTHY 持续多久后可由 Controller 回收；`0s`
（默认）表示只保留异常状态，永不因该策略删除。首次进入 UNHEALTHY 的时间会跨
CHECKING 重试保持不变，成功恢复 AVAILABLE 时清除。
- 回收器只运行在 Controller 后台，使用 Redis 有序索引和固定批次，不进入 Gateway
请求热路径。当前只原子删除没有 Worker ownership 的 Proxy；仍属于 Worker 的候选会
延后，等待既有 Drain/ACK 完成后再由下一轮回收，避免删除仍可能出现在 Gateway
Snapshot 中的代理。
## 9. 存储、Admin 与 Metrics

8
docs/development/implementation-plan.md
View File

@ -262,7 +262,8 @@ Gateway 请求热路径仍只使用本地准入，不增加 Redis/PostgreSQL 调
WorkerControlPlane gRPC 接收端、session 签发/心跳、Snapshot ACK 账本、基础
Snapshot 流、Gateway 会话客户端与快照凭据分发已完成；权威 Proxy/Routing 发布、
Outcome 上报已完成为有界队列、批次序列/摘要栅栏和确认重试；健康 BASIC/EGRESS/TARGET
执行链已完成，REMOVE 生命周期仍待完成，因此 Task 10 尚未全部完成。
执行链已完成；持续 UNHEALTHY 的未分配 Proxy 可由后台有界回收，拥有 Worker
所有权的项仍等待完整 Drain/ACK 触发，不直接删除，因此 Task 10 尚未全部完成。
## Task 11: Checker and Health Reducer
@ -290,7 +291,10 @@ EGRESS 的出口身份响应解析支持固定上限的纯文本和常见 JSON I
`routing.check.targets` 作为配置入口，限定每个 Routing 的 URL 数和每个 Upstream 的 Profile
总数；Redis 为 `(routing_name, target_url, proxy_id)` 维护独立 due-index，完成、租约回收和
代理过期均沿用同一原子任务边界。BASIC、EGRESS 和 TARGET 分组轮转并共享 Upstream in-flight
上限，因此本任务已完成。
上限，因此本任务已完成。持续 UNHEALTHY 回收使用每 Upstream 的
`check.unhealthyRemoveAfter`：全局 Reducer 保存首次进入 UNHEALTHY 的时间，Redis 以
有序索引有界扫描；只有无 Worker ownership 的项会沿用统一清理边界删除，已拥有
Worker 的项延后至 Drain/ACK 解除所有权后重试。
补充进度（2026-08-02）：BASIC 调度已改为配置驱动监督器。它每轮读取已发布快照并复用
有界派发逻辑，所以 reload 后已启用上游的策略变更、停用，以及新启用上游都无需重启

11
docs/operations/runbook.md
View File

@ -296,6 +296,17 @@ Outbox 发布器必须以稳定 consumer ID 有界领取；发布成功后原子
3. 检查目标超时、DNS 与出口网络，再按任务延迟扩容 Checker。
4. 队列必须有上限；不得无限积压耗尽内存或 Redis。
### 7.8 持续 UNHEALTHY 回收
1. 先确认 `check.maxConsecutiveFailures`、检查目标和出口网络；首次失败进入
SUSPECT，阈值达到后才进入 UNHEALTHY。
2. `check.unhealthyRemoveAfter: 0s` 仅保留异常状态；设置正值后 Controller 才会在
持续异常超过该窗口时执行有界回收。
3. 回收器只删除没有 Worker ownership 的记录。发现所有权时会延后处理，不得手动
删除 Redis 记录或索引。
4. 对已分配代理，按既有流程执行 DRAINING，等待 Worker ACK 且 active/reserved
都归零；所有权清除后，下一轮回收才会删除该异常 Proxy。
## 8. 备份与恢复
- PostgreSQL：每日全量、连续 WAL/PITR，保护配置版本、Upstream/Routing 管理

2
docs/requirements/traceability.md
View File

@ -81,7 +81,7 @@
|---|---|---|---|
| HEALTH-001 | 全局健康与 Routing/目标健康分离 | 221-270, 8679-8708 | `domain/health` 已将 BASIC/EGRESS 全局 Reducer 与 TARGET Profile Reducer 分离；TARGET 在 Memory 和 Redis 独立、随代理 TTL 归并，不改写 Proxy 全局状态，并按 `(routing_name, target_url)` 进入生产调度 |
| HEALTH-002 | 健康调度有 jitter、maxInFlight 和分级频率 | 8679-8736 | 配置有效合并、URL/Profile 上限、稳定抖动/优先级 Planner、有界 Scheduler tick、Redis BASIC/EGRESS/TARGET due-index、跨副本 in-flight 原子限制，以及 HTTP/HTTPS/SOCKS5 BASIC/EGRESS/TARGET 生产执行器已完成 |
| HEALTH-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | Controller 公用 Reducer 已通过 Memory/Redis 活动池原子提交全局连续失败、精确重放和成功恢复；BASIC 任务调度已完成，REMOVE 编排待实现 |
| HEALTH-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | Controller 公用 Reducer 已通过 Memory/Redis 活动池原子提交全局连续失败、精确重放和成功恢复；`check.unhealthyRemoveAfter` 驱动后台有界回收，Redis/Memory 复用相同窄接口并只删除无 Worker ownership 的异常项。已拥有 Worker 的项会延后到 Drain/ACK 清除所有权后重试，自动触发 Drain 的发布闭环仍待完成 |
| SEC-001 | API 认证与 Proxy 认证分离，Secret 统一脱敏 | 7528-8111, 8904-8945 | Config 脱敏、Provider Store -> SecretRef -> Gateway Resolver 跨包测试与格式化泄漏回归测试 |
| SEC-002 | 非回环监听无保护时严格模式启动失败 | 8112-8441 | 配置校验测试 |
| OPS-001 | 配置校验后构建不可变快照并原子替换 | 8959-8999 | 100k 索引、版本/epoch 与并发 Apply/Acquire 测试 |

14
findings.md
View File

@ -100,6 +100,20 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两
HealthObservation，后续 Go 领域类型必须保持字段语义一致，但不直接依赖生成的
transport 类型。
## 健康故障移除实现（2026-08-02）
- `health.GlobalState` 和 Redis 活动记录都保存首次 `UNHEALTHY` 时间；后续
`UNHEALTHY -> CHECKING -> UNHEALTHY` 复检不会重置，成功恢复 `AVAILABLE` 才清除。
- `check.unhealthyRemoveAfter` 默认 `0s`，按 Effective Check 为每个启用 Upstream
生效。Controller Reaper 读取当前配置快照，以固定批次调用公用
`activitypool.UnhealthyRemover`，不进入 Gateway 热路径。
- Redis 使用 `{activity}` 槽内的有序索引扫描到期候选；Memory 参考实现保持相同
`SweepUnhealthy` 契约。两者只删除无 Worker ownership 的记录，仍归属 Worker 的
候选延后一秒后再检查，不自行发起 Drain。
- Health、Upsert、Extract、Ownership、状态读取与过期清理脚本都在同一原子删除边界
维护该索引。Memory 单测、Redis 集成契约和配置/Reaper 单测覆盖恢复、阈值、延后与
Drain/ACK 后清理。
## Git 同步事实（2026-07-29）
- PostgreSQL 管理面基础文档已提交为 `7951c29`。

5
internal/adapters/redisactivity/codec.go
View File

@ -33,6 +33,7 @@ type proxyRecord struct {
LastCheckedAtMS int64 `json:"lastCheckedAtMs,omitempty"`
LastSuccessAtMS int64 `json:"lastSuccessAtMs,omitempty"`
ConsecutiveFailures int `json:"consecutiveFailures,omitempty"`
UnhealthySinceMS int64 `json:"unhealthySinceMs,omitempty"`
LastHealthTaskID string `json:"lastHealthTaskId,omitempty"`
LastHealthDigest string `json:"lastHealthDigest,omitempty"`
LastHealthObservedAtMS int64 `json:"lastHealthObservedAtMs,omitempty"`
@ -213,7 +214,9 @@ func validateProxyRecord(record proxyRecord) error {
record.OwnerIndexKey == "" || !strings.Contains(record.OwnerIndexKey, "{activity}") {
return ErrInvalidRecord
}
if (record.LastHealthTaskID == "") != (record.LastHealthDigest == "") ||
if record.UnhealthySinceMS < 0 || (record.UnhealthySinceMS > 0 && record.State != "UNHEALTHY" && record.State != "CHECKING") ||
(record.State == "UNHEALTHY" && record.UnhealthySinceMS == 0) ||
(record.LastHealthTaskID == "") != (record.LastHealthDigest == "") ||
(record.LastHealthDigest != "" && (!validDigest(record.LastHealthDigest) || record.LastHealthObservedAtMS <= 0)) ||
record.LastHealthObservedAtMS < 0 {
return ErrInvalidRecord

2
internal/adapters/redisactivity/extract.go
View File

@ -91,7 +91,7 @@ func (a *Adapter) Extract(ctx context.Context, command extractionDomain.Command)
}
keys = append(keys, a.extractionDriverKeys(digestInput)...)
keys = append(keys, a.keys.healthDue, a.keys.healthQueued, a.keys.healthLeases,
a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask)
a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask, a.keys.healthUnhealthy)
idempotencyTTL := command.IdempotencyTTL
if idempotencyTTL == 0 {
idempotencyTTL = defaultRedisIdempotencyTTL

97
internal/adapters/redisactivity/global_health_integration_test.go
View File

@ -8,6 +8,8 @@ import (
"testing"
"time"
"github.com/redis/go-redis/v9"
"proxy-pool/internal/domain/activitypool"
healthDomain "proxy-pool/internal/domain/health"
proxyDomain "proxy-pool/internal/domain/proxy"
@ -31,7 +33,8 @@ func TestRedisAppliesGlobalHealthObservationAtomically(t *testing.T) {
failed, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
Observation: first, MaxConsecutiveFailures: 3,
})
if err != nil || failed.State != proxyDomain.StateUnhealthy || failed.GlobalHealth.ConsecutiveFailures != 1 {
if err != nil || failed.State != proxyDomain.StateUnhealthy || failed.GlobalHealth.ConsecutiveFailures != 1 ||
!failed.GlobalHealth.UnhealthySince.Equal(now.Add(2*time.Second)) {
t.Fatalf("ApplyGlobalObservation(initial failure) = %+v, %v", failed, err)
}
replayed, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
@ -57,11 +60,103 @@ func TestRedisAppliesGlobalHealthObservationAtomically(t *testing.T) {
Observation: redisGlobalObservation("task-2", true, now.Add(4*time.Second)), MaxConsecutiveFailures: 3,
})
if err != nil || recovered.State != proxyDomain.StateAvailable || recovered.GlobalHealth.ConsecutiveFailures != 0 ||
!recovered.GlobalHealth.UnhealthySince.IsZero() ||
recovered.GlobalHealth.LastObservedAt != now.Add(4*time.Second) {
t.Fatalf("ApplyGlobalObservation(recovery) = %+v, %v", recovered, err)
}
}
func TestRedisSweepsUnownedProxiesAfterUnhealthyGrace(t *testing.T) {
fixture := newRedisTestFixture(t)
now := redisTestNow()
if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1,
Proxies: []proxyDomain.Proxy{testProxy("proxy-a", "192.0.2.10")},
}); err != nil {
t.Fatalf("UpsertFetched() = %v", err)
}
if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{
ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking,
}); err != nil {
t.Fatalf("ApplyHealth(checking) = %v", err)
}
if _, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
Observation: redisGlobalObservation("task-unhealthy", false, now.Add(2*time.Second)), MaxConsecutiveFailures: 2,
}); err != nil {
t.Fatalf("ApplyGlobalObservation(unhealthy) = %v", err)
}
if _, err := fixture.Client.ZScore(context.Background(), fixture.Adapter.keys.healthUnhealthy, "proxy-a").Result(); err != nil {
t.Fatalf("health unhealthy index = %v", err)
}
command := activitypool.UnhealthySweepCommand{
Now: now.Add(3 * time.Second), Limit: 4, RemoveAfterByUpstream: map[string]time.Duration{"provider-a": 3 * time.Second},
}
if result, err := fixture.Adapter.SweepUnhealthy(context.Background(), command); err != nil || result != (activitypool.UnhealthySweepResult{}) {
t.Fatalf("SweepUnhealthy(before grace) = %+v, %v", result, err)
}
command.Now = now.Add(6 * time.Second)
if result, err := fixture.Adapter.SweepUnhealthy(context.Background(), command); err != nil || result != (activitypool.UnhealthySweepResult{Removed: 1}) {
t.Fatalf("SweepUnhealthy(after grace) = %+v, %v", result, err)
}
if _, err := fixture.Adapter.UpstreamForProxy(context.Background(), "proxy-a", command.Now); !errors.Is(err, activitypool.ErrActivityNotFound) {
t.Fatalf("UpstreamForProxy(removed) error = %v, want ErrActivityNotFound", err)
}
if _, err := fixture.Client.ZScore(context.Background(), fixture.Adapter.keys.healthUnhealthy, "proxy-a").Result(); !errors.Is(err, redis.Nil) {
t.Fatalf("health unhealthy index after removal = %v, want redis.Nil", err)
}
}
func TestRedisUnhealthySweepDefersOwnedProxyUntilDrainAcknowledgement(t *testing.T) {
fixture := newRedisTestFixture(t)
now := redisTestNow()
if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1,
Proxies: []proxyDomain.Proxy{testProxy("proxy-a", "192.0.2.10")},
}); err != nil {
t.Fatalf("UpsertFetched() = %v", err)
}
if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{
ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking,
}); err != nil {
t.Fatalf("ApplyHealth(checking) = %v", err)
}
if _, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
Observation: redisGlobalObservation("task-available", true, now.Add(2*time.Second)), MaxConsecutiveFailures: 2,
}); err != nil {
t.Fatalf("ApplyGlobalObservation(available) = %v", err)
}
assignment, err := fixture.Adapter.Assign(context.Background(), now.Add(3*time.Second), "proxy-a", "worker-a", time.Minute)
if err != nil {
t.Fatalf("Assign() = %v", err)
}
if _, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
Observation: redisGlobalObservation("task-suspect", false, now.Add(4*time.Second)), MaxConsecutiveFailures: 2,
}); err != nil {
t.Fatalf("ApplyGlobalObservation(suspect) = %v", err)
}
if _, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{
Observation: redisGlobalObservation("task-unhealthy", false, now.Add(5*time.Second)), MaxConsecutiveFailures: 2,
}); err != nil {
t.Fatalf("ApplyGlobalObservation(unhealthy) = %v", err)
}
command := activitypool.UnhealthySweepCommand{
Now: now.Add(7 * time.Second), Limit: 4, RemoveAfterByUpstream: map[string]time.Duration{"provider-a": time.Second},
}
if result, err := fixture.Adapter.SweepUnhealthy(context.Background(), command); err != nil || result != (activitypool.UnhealthySweepResult{DeferredOwned: 1}) {
t.Fatalf("SweepUnhealthy(owned) = %+v, %v", result, err)
}
if _, err := fixture.Adapter.BeginDrain(context.Background(), "proxy-a", "worker-a", assignment.Epoch); err != nil {
t.Fatalf("BeginDrain() = %v", err)
}
if err := fixture.Adapter.AcknowledgeDrain(context.Background(), "proxy-a", "worker-a", assignment.Epoch, 0, 0); err != nil {
t.Fatalf("AcknowledgeDrain() = %v", err)
}
command.Now = now.Add(8 * time.Second)
if result, err := fixture.Adapter.SweepUnhealthy(context.Background(), command); err != nil || result != (activitypool.UnhealthySweepResult{Removed: 1}) {
t.Fatalf("SweepUnhealthy(unowned) = %+v, %v", result, err)
}
}
func redisGlobalObservation(taskID string, success bool, observedAt time.Time) healthDomain.Observation {
observation := healthDomain.Observation{
TaskID: taskID, ProxyID: "proxy-a", Level: healthDomain.LevelBasic,

5
internal/adapters/redisactivity/health.go
View File

@ -14,6 +14,7 @@ import (
var _ activitypool.HealthStore = (*Adapter)(nil)
var _ activitypool.GlobalHealthStore = (*Adapter)(nil)
var _ activitypool.TargetHealthStore = (*Adapter)(nil)
var _ activitypool.UnhealthyRemover = (*Adapter)(nil)
type targetHealthRecord struct {
Version int `json:"version"`
@ -45,7 +46,7 @@ func (a *Adapter) ApplyHealth(ctx context.Context, update activitypool.HealthUpd
a.keys.records, a.keys.unique, a.keys.idkeys, a.keys.expiry, a.keys.available,
a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry,
a.keys.operation(operationID), a.keys.healthDue, a.keys.healthQueued, a.keys.healthLeases,
a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask,
a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask, a.keys.healthUnhealthy,
}, update.CheckedAt.UnixMilli(), string(update.NextState), int64(update.Latency),
a.options.CleanupLimit, operationTTLMillis(a.options.OperationTTL), update.ProxyID)
if err != nil {
@ -111,7 +112,7 @@ func (a *Adapter) ApplyGlobalObservation(
a.keys.records, a.keys.unique, a.keys.idkeys, a.keys.expiry, a.keys.available,
a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry,
a.keys.operation(operationID), a.keys.healthDue, a.keys.healthQueued, a.keys.healthLeases,
a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask,
a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask, a.keys.healthUnhealthy,
}, normalized.ObservedAt.UnixMilli(), "", int64(normalized.Latency),
a.options.CleanupLimit, operationTTLMillis(a.options.OperationTTL), normalized.ProxyID,
"global", success, command.MaxConsecutiveFailures, normalized.TaskID, hex.EncodeToString(digest[:]))

2
internal/adapters/redisactivity/keys.go
View File

@ -28,6 +28,7 @@ type keyspace struct {
workerRuntimeExpiry string
workerOutcomes string
healthDue string
healthUnhealthy string
healthEgressDue string
healthTargetDue string
healthQueued string
@ -59,6 +60,7 @@ func newKeyspace(namespace string) keyspace {
workerRuntimeExpiry: prefix + ":worker-runtime-expiry",
workerOutcomes: prefix + ":worker-outcomes",
healthDue: prefix + ":health-due",
healthUnhealthy: prefix + ":health-unhealthy",
healthEgressDue: prefix + ":health-egress-due",
healthTargetDue: prefix + ":health-target-due",
healthQueued: prefix + ":health-queued",

58
internal/adapters/redisactivity/maintenance.go
View File

@ -2,6 +2,7 @@ package redisactivity
import (
"context"
"encoding/json"
"time"
"proxy-pool/internal/domain/activitypool"
@ -10,6 +11,9 @@ import (
const (
maintenanceInventory = "inventory"
maintenanceSweep = "sweep"
maintenanceUnhealthy = "unhealthy"
maximumUnhealthySweepScan = 1024
)
var (
@ -28,7 +32,7 @@ func (a *Adapter) Inventory(ctx context.Context, upstreamID string, now time.Tim
if a == nil || upstreamID == "" || now.IsZero() {
return result, activitypool.ErrInvalidInventory
}
reply, err := a.runMaintenance(ctx, maintenanceInventory, now, a.options.CleanupLimit, upstreamID)
reply, err := a.runMaintenance(ctx, maintenanceInventory, now, a.options.CleanupLimit, upstreamID, "")
if err != nil {
return result, err
}
@ -52,7 +56,7 @@ func (a *Adapter) SweepExpired(ctx context.Context, now time.Time, limit int) (i
if a == nil || now.IsZero() || limit <= 0 {
return 0, activitypool.ErrInvalidMaintenance
}
reply, err := a.runMaintenance(ctx, maintenanceSweep, now, limit, "")
reply, err := a.runMaintenance(ctx, maintenanceSweep, now, limit, "", "")
if err != nil {
return 0, err
}
@ -65,12 +69,58 @@ func (a *Adapter) SweepExpired(ctx context.Context, now time.Time, limit int) (i
return reply.Count, nil
}
func (a *Adapter) SweepUnhealthy(
ctx context.Context,
command activitypool.UnhealthySweepCommand,
) (activitypool.UnhealthySweepResult, error) {
if ctx == nil {
return activitypool.UnhealthySweepResult{}, activitypool.ErrInvalidMaintenance
}
if err := ctx.Err(); err != nil {
return activitypool.UnhealthySweepResult{}, err
}
if a == nil || command.Now.IsZero() || command.Limit <= 0 || len(command.RemoveAfterByUpstream) == 0 {
return activitypool.UnhealthySweepResult{}, activitypool.ErrInvalidMaintenance
}
policies := make(map[string]int64, len(command.RemoveAfterByUpstream))
for upstreamID, removeAfter := range command.RemoveAfterByUpstream {
if upstreamID == "" || removeAfter <= 0 {
return activitypool.UnhealthySweepResult{}, activitypool.ErrInvalidMaintenance
}
policies[upstreamID] = removeAfter.Milliseconds()
if policies[upstreamID] <= 0 {
return activitypool.UnhealthySweepResult{}, activitypool.ErrInvalidMaintenance
}
}
payload, err := json.Marshal(policies)
if err != nil {
return activitypool.UnhealthySweepResult{}, err
}
limit := command.Limit
if limit > maximumUnhealthySweepScan/4 {
limit = maximumUnhealthySweepScan / 4
}
reply, err := a.runMaintenance(ctx, maintenanceUnhealthy, command.Now, limit, "", string(payload))
if err != nil {
return activitypool.UnhealthySweepResult{}, err
}
if reply.Status == scriptInvalid {
return activitypool.UnhealthySweepResult{}, activitypool.ErrInvalidMaintenance
}
if reply.Status != scriptOK || reply.Count < 0 || reply.Count > limit || reply.DeferredOwned < 0 ||
reply.DeferredOwned > maximumUnhealthySweepScan {
return activitypool.UnhealthySweepResult{}, invalidScriptReply("unexpected unhealthy sweep reply")
}
return activitypool.UnhealthySweepResult{Removed: reply.Count, DeferredOwned: reply.DeferredOwned}, nil
}
func (a *Adapter) runMaintenance(
ctx context.Context,
operation string,
now time.Time,
limit int,
upstreamID string,
policies string,
) (maintenanceScriptReply, error) {
operationID, err := newOperationID()
if err != nil {
@ -80,8 +130,8 @@ func (a *Adapter) runMaintenance(
a.keys.records, a.keys.unique, a.keys.idkeys, a.keys.expiry, a.keys.available,
a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry,
a.keys.operation(operationID), a.keys.healthDue, a.keys.healthQueued, a.keys.healthLeases,
a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask,
}, operation, now.UnixMilli(), limit, upstreamID, operationTTLMillis(a.options.OperationTTL))
a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask, a.keys.healthUnhealthy,
}, operation, now.UnixMilli(), limit, upstreamID, operationTTLMillis(a.options.OperationTTL), policies)
if err != nil {
return maintenanceScriptReply{}, err
}

2
internal/adapters/redisactivity/ownership.go
View File

@ -225,7 +225,7 @@ func (a *Adapter) runOwnership(
a.keys.records, a.keys.unique, a.keys.idkeys, a.keys.expiry, a.keys.available,
a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry,
a.keys.epoch, operationKey, a.keys.healthDue, a.keys.healthQueued, a.keys.healthLeases,
a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask,
a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask, a.keys.healthUnhealthy,
}, operation, operationTTLMillis(a.options.OperationTTL), a.options.CleanupLimit,
nowMS, proxyID, workerID, epoch, value, active, reserved, a.keys.workerOwned(workerID))
if err != nil {

8
internal/adapters/redisactivity/runtime_contract_integration_test.go
View File

@ -11,13 +11,13 @@ import (
)
func TestRedisWorkerControlStoreContract(t *testing.T) {
contracttest.Run(t, func(*testing.T) contracttest.Fixture {
fixture := newRedisTestFixture(t)
contracttest.Run(t, func(test *testing.T) contracttest.Fixture {
fixture := newRedisTestFixture(test)
now := redisTestNow()
seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), 2*time.Minute,
seedRedisAvailable(test, fixture.Adapter, "provider-a", now, now.Add(time.Second), 2*time.Minute,
testProxy("proxy-a", "192.0.2.10"))
if _, err := fixture.Adapter.Assign(context.Background(), now.Add(2*time.Second), "proxy-a", "worker-a", time.Minute); err != nil {
t.Fatalf("Assign(): %v", err)
test.Fatalf("Assign(): %v", err)
}
return contracttest.Fixture{
Store: fixture.Adapter, Reader: fixture.Adapter, Outcomes: fixture.Adapter, TTL: 100 * time.Millisecond,

1
internal/adapters/redisactivity/scripts.go
View File

@ -90,6 +90,7 @@ type ownershipScriptReply struct {
type maintenanceScriptReply struct {
Status scriptStatus `json:"status"`
Count int `json:"count"`
DeferredOwned int `json:"deferredOwned"`
}
type statusScriptReply struct {

14
internal/adapters/redisactivity/scripts/extract.lua
View File

@ -9,12 +9,13 @@ local owners_key = KEYS[8]
local owner_expiry_key = KEYS[9]
local operation_key = KEYS[10]
local idempotency_key = KEYS[11]
local health_due_key = KEYS[#KEYS - 5]
local health_queued_key = KEYS[#KEYS - 4]
local health_leases_key = KEYS[#KEYS - 3]
local health_tasks_key = KEYS[#KEYS - 2]
local health_task_expiry_key = KEYS[#KEYS - 1]
local health_ref_task_key = KEYS[#KEYS]
local health_due_key = KEYS[#KEYS - 6]
local health_queued_key = KEYS[#KEYS - 5]
local health_leases_key = KEYS[#KEYS - 4]
local health_tasks_key = KEYS[#KEYS - 3]
local health_task_expiry_key = KEYS[#KEYS - 2]
local health_ref_task_key = KEYS[#KEYS - 1]
local health_unhealthy_key = KEYS[#KEYS]
local now_ms = tonumber(ARGV[1])
local requested = tonumber(ARGV[2])
@ -189,6 +190,7 @@ local function remove_proxy(proxy_id)
remove_worker_owned(proxy_id)
redis.call('HDEL', owners_key, proxy_id)
redis.call('ZREM', owner_expiry_key, proxy_id)
redis.call('ZREM', health_unhealthy_key, proxy_id)
remove_health_task(proxy_id)
end

13
internal/adapters/redisactivity/scripts/health.lua
View File

@ -14,6 +14,7 @@ local health_leases_key = KEYS[13]
local health_tasks_key = KEYS[14]
local health_task_expiry_key = KEYS[15]
local health_ref_task_key = KEYS[16]
local health_unhealthy_key = KEYS[17]
local checked_at_ms = tonumber(ARGV[1])
local next_state = ARGV[2]
@ -171,6 +172,7 @@ local function remove_proxy(id)
remove_worker_owned(id)
redis.call('HDEL', owners_key, id)
redis.call('ZREM', owner_expiry_key, id)
redis.call('ZREM', health_unhealthy_key, id)
remove_health_task(id)
end
@ -301,6 +303,17 @@ record.latencyNs = latency_ns
if next_state == 'AVAILABLE' then
record.lastSuccessAtMs = checked_at_ms
end
if next_state == 'UNHEALTHY' then
local unhealthy_since_ms = tonumber(record.unhealthySinceMs or '0')
if not unhealthy_since_ms or unhealthy_since_ms <= 0 then
unhealthy_since_ms = checked_at_ms
end
record.unhealthySinceMs = unhealthy_since_ms
redis.call('ZADD', health_unhealthy_key, unhealthy_since_ms, proxy_id)
else
record.unhealthySinceMs = nil
redis.call('ZREM', health_unhealthy_key, proxy_id)
end
if was_managed and not will_be_managed then
decrement_inventory(record.sourceUpstream)
elseif not was_managed and will_be_managed then

2
internal/adapters/redisactivity/scripts/ownership.lua
View File

@ -15,6 +15,7 @@ local health_leases_key = KEYS[14]
local health_tasks_key = KEYS[15]
local health_task_expiry_key = KEYS[16]
local health_ref_task_key = KEYS[17]
local health_unhealthy_key = KEYS[18]
local operation = ARGV[1]
local operation_ttl_ms = tonumber(ARGV[2])
@ -175,6 +176,7 @@ local function remove_proxy(id)
redis.call('ZREM', expiry_key, id)
redis.call('HDEL', owners_key, id)
redis.call('ZREM', owner_expiry_key, id)
redis.call('ZREM', health_unhealthy_key, id)
remove_health_task(id)
end

2
internal/adapters/redisactivity/scripts/status.lua
View File

@ -13,6 +13,7 @@ local health_leases_key = KEYS[12]
local health_tasks_key = KEYS[13]
local health_task_expiry_key = KEYS[14]
local health_ref_task_key = KEYS[15]
local health_unhealthy_key = KEYS[16]
local now_ms = tonumber(ARGV[1])
local cleanup_limit = tonumber(ARGV[2])
@ -115,6 +116,7 @@ local function remove_proxy(proxy_id)
redis.call('ZREM', expiry_key, proxy_id)
redis.call('HDEL', owners_key, proxy_id)
redis.call('ZREM', owner_expiry_key, proxy_id)
redis.call('ZREM', health_unhealthy_key, proxy_id)
remove_health_task(proxy_id)
end

44
internal/adapters/redisactivity/scripts/sweep.lua
View File

@ -14,12 +14,14 @@ local health_leases_key = KEYS[13]
local health_tasks_key = KEYS[14]
local health_task_expiry_key = KEYS[15]
local health_ref_task_key = KEYS[16]
local health_unhealthy_key = KEYS[17]
local operation = ARGV[1]
local now_ms = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])
local upstream_id = ARGV[4]
local operation_ttl_ms = tonumber(ARGV[5])
local unhealthy_policies_json = ARGV[6]
local function finish(reply)
local encoded = cjson.encode(reply)
@ -149,6 +151,7 @@ local function remove_proxy(proxy_id)
remove_worker_owned(proxy_id)
redis.call('HDEL', owners_key, proxy_id)
redis.call('ZREM', owner_expiry_key, proxy_id)
redis.call('ZREM', health_unhealthy_key, proxy_id)
remove_health_task(proxy_id)
end
@ -160,6 +163,47 @@ end
if operation == 'sweep' then
return finish({status = 'ok', count = #expired})
end
if operation == 'unhealthy' then
local decoded, policies = pcall(cjson.decode, unhealthy_policies_json or '')
if not decoded or type(policies) ~= 'table' then
return finish({status = 'invalid', count = 0, deferredOwned = 0})
end
local scan_limit = limit * 4
if scan_limit > 1024 then
scan_limit = 1024
end
local candidates = redis.call('ZRANGEBYSCORE', health_unhealthy_key, '-inf', now_ms, 'LIMIT', 0, scan_limit)
local removed = 0
local deferred_owned = 0
for _, proxy_id in ipairs(candidates) do
local raw = redis.call('HGET', records_key, proxy_id)
if not raw then
redis.call('ZREM', health_unhealthy_key, proxy_id)
else
local valid, record = pcall(cjson.decode, raw)
local remove_after_ms = valid and type(record) == 'table' and tonumber(policies[record.sourceUpstream]) or nil
local unhealthy_since_ms = valid and type(record) == 'table' and tonumber(record.unhealthySinceMs or '0') or nil
if not valid or type(record) ~= 'table' or record.state ~= 'UNHEALTHY' or not remove_after_ms or
remove_after_ms <= 0 or not unhealthy_since_ms or unhealthy_since_ms <= 0 then
redis.call('ZREM', health_unhealthy_key, proxy_id)
elseif unhealthy_since_ms + remove_after_ms > now_ms then
redis.call('ZADD', health_unhealthy_key, unhealthy_since_ms + remove_after_ms, proxy_id)
else
local owner_raw = redis.call('HGET', owners_key, proxy_id)
local owned = owner_raw or (type(record.ownerWorkerId) == 'string' and record.ownerWorkerId ~= '')
if owned then
local retry_at_ms = now_ms + 1000
redis.call('ZADD', health_unhealthy_key, retry_at_ms, proxy_id)
deferred_owned = deferred_owned + 1
elseif removed < limit then
remove_proxy(proxy_id)
removed = removed + 1
end
end
end
end
return finish({status = 'ok', count = removed, deferredOwned = deferred_owned})
end
if operation == 'inventory' then
local count = tonumber(redis.call('HGET', inventory_key, upstream_id) or '0')
if count < 0 then

3
internal/adapters/redisactivity/scripts/upsert.lua
View File

@ -14,6 +14,7 @@ local health_leases_key = KEYS[13]
local health_tasks_key = KEYS[14]
local health_task_expiry_key = KEYS[15]
local health_ref_task_key = KEYS[16]
local health_unhealthy_key = KEYS[17]
local now_ms = tonumber(ARGV[1])
local cleanup_limit = tonumber(ARGV[2])
@ -141,6 +142,7 @@ local function remove_proxy(proxy_id)
remove_worker_owned(proxy_id)
redis.call('HDEL', owners_key, proxy_id)
redis.call('ZREM', owner_expiry_key, proxy_id)
redis.call('ZREM', health_unhealthy_key, proxy_id)
redis.call('ZREM', health_due_key, proxy_id)
local task_id = redis.call('HGET', health_ref_task_key, proxy_id)
if task_id then
@ -255,6 +257,7 @@ for _, candidate in ipairs(candidates) do
incoming.lastCheckedAtMs = current.lastCheckedAtMs
incoming.lastSuccessAtMs = current.lastSuccessAtMs
incoming.consecutiveFailures = current.consecutiveFailures
incoming.unhealthySinceMs = current.unhealthySinceMs
incoming.lastHealthTaskId = current.lastHealthTaskId
incoming.lastHealthDigest = current.lastHealthDigest
incoming.lastHealthObservedAtMs = current.lastHealthObservedAtMs

2
internal/adapters/redisactivity/status.go
View File

@ -40,7 +40,7 @@ func (a *Adapter) ReadStateInventory(
a.keys.records, a.keys.unique, a.keys.idkeys, a.keys.expiry, a.keys.available,
a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry,
a.keys.healthDue, a.keys.healthQueued, a.keys.healthLeases, a.keys.healthTasks,
a.keys.healthTaskExpiry, a.keys.healthRefTask,
a.keys.healthTaskExpiry, a.keys.healthRefTask, a.keys.healthUnhealthy,
}, now.UnixMilli(), a.options.CleanupLimit, string(payload))
if err != nil {
return nil, err

4
internal/adapters/redisactivity/upsert.go
View File

@ -206,6 +206,7 @@ func (a *Adapter) upsertChunk(
a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry,
a.keys.operation(operationID), a.keys.healthDue, a.keys.healthQueued,
a.keys.healthLeases, a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask,
a.keys.healthUnhealthy,
}, observedAt.UnixMilli(), a.options.CleanupLimit, maxSize, operationTTLMillis(a.options.OperationTTL), string(payload))
if err != nil {
return upsertScriptReply{}, err
@ -255,6 +256,9 @@ func proxyRecordEntry(record proxyRecord) activitypool.Entry {
proxy.LastSuccessAt = &value
}
global := healthDomain.GlobalState{State: proxyDomain.State(record.State)}
if record.UnhealthySinceMS > 0 {
global.UnhealthySince = time.UnixMilli(record.UnhealthySinceMS).UTC()
}
if record.LastHealthTaskID != "" {
digest, err := hex.DecodeString(record.LastHealthDigest)
if err == nil && len(digest) == sha256.Size && record.LastHealthObservedAtMS > 0 {

1
internal/config/config.go
View File

@ -314,5 +314,6 @@ type Check struct {
Timeout Duration `yaml:"timeout"`
MaxAttempts int `yaml:"maxAttempts"`
MaxConsecutiveFailures int `yaml:"maxConsecutiveFailures"`
UnhealthyRemoveAfter Duration `yaml:"unhealthyRemoveAfter"`
URLs []string `yaml:"urls"`
}

7
internal/config/config_test.go
View File

@ -816,6 +816,13 @@ func TestValidateRejectsInvalidConfigurationMatrix(t *testing.T) {
},
want: "maxConsecutiveFailures",
},
{
name: "negative unhealthy remove after",
mutate: func(cfg *Config) {
updateUpstream(cfg, func(upstream *Upstream) { upstream.Check.UnhealthyRemoveAfter = Duration(-time.Second) })
},
want: "unhealthyRemoveAfter",
},
}
for _, test := range tests {

3
internal/config/effective_check.go
View File

@ -23,6 +23,9 @@ func EffectiveCheck(defaults, override Check) Check {
if override.MaxConsecutiveFailures != 0 {
effective.MaxConsecutiveFailures = override.MaxConsecutiveFailures
}
if override.UnhealthyRemoveAfter != 0 {
effective.UnhealthyRemoveAfter = override.UnhealthyRemoveAfter
}
if override.URLs != nil {
effective.URLs = cloneCheckURLs(override.URLs)
} else {

8
internal/config/effective_check_test.go
View File

@ -10,7 +10,8 @@ import (
func TestEffectiveCheckOverlaysScalarsAndClonesURLs(t *testing.T) {
defaults := Check{
Interval: Duration(time.Minute), Jitter: 20, MaxInFlight: 12, Timeout: Duration(5 * time.Second),
MaxAttempts: 3, MaxConsecutiveFailures: 2, URLs: []string{"https://egress.example/check"},
MaxAttempts: 3, MaxConsecutiveFailures: 2, UnhealthyRemoveAfter: Duration(3 * time.Minute),
URLs: []string{"https://egress.example/check"},
}
override := Check{Timeout: Duration(8 * time.Second), MaxAttempts: 5}
effective := EffectiveCheck(defaults, override)
@ -18,6 +19,7 @@ func TestEffectiveCheckOverlaysScalarsAndClonesURLs(t *testing.T) {
effective.MaxInFlight != defaults.MaxInFlight || effective.Timeout != override.Timeout ||
effective.MaxAttempts != override.MaxAttempts ||
effective.MaxConsecutiveFailures != defaults.MaxConsecutiveFailures ||
effective.UnhealthyRemoveAfter != defaults.UnhealthyRemoveAfter ||
len(effective.URLs) != 1 || effective.URLs[0] != defaults.URLs[0] {
t.Fatalf("EffectiveCheck() = %+v", effective)
}
@ -30,6 +32,10 @@ func TestEffectiveCheckOverlaysScalarsAndClonesURLs(t *testing.T) {
if effective.URLs == nil || len(effective.URLs) != 0 {
t.Fatalf("EffectiveCheck(explicit empty URLs) = %+v", effective)
}
effective = EffectiveCheck(defaults, Check{UnhealthyRemoveAfter: Duration(5 * time.Minute)})
if effective.UnhealthyRemoveAfter != Duration(5*time.Minute) {
t.Fatalf("EffectiveCheck(remove after override) = %+v", effective)
}
}
func TestValidateUsesEffectiveCheckAndRejectsUnsafeURLs(t *testing.T) {

4
internal/config/validate.go
View File

@ -608,6 +608,9 @@ func validateCheck(scope string, check Check) error {
if err := requirePositive(scope+".maxConsecutiveFailures", check.MaxConsecutiveFailures); err != nil {
return err
}
if err := requireNonNegative(scope+".unhealthyRemoveAfter", check.UnhealthyRemoveAfter); err != nil {
return err
}
return validateCheckURLs(scope+".urls", check.URLs)
}
@ -656,6 +659,7 @@ func fetchConfigured(fetch Fetch) bool {
func checkConfigured(check Check) bool {
return check.Interval != 0 || check.Jitter != 0 || check.MaxInFlight != 0 ||
check.Timeout != 0 || check.MaxAttempts != 0 || check.MaxConsecutiveFailures != 0 ||
check.UnhealthyRemoveAfter != 0 ||
len(check.URLs) != 0
}

13
internal/controller/bootstrap/bootstrap.go
View File

@ -321,6 +321,19 @@ func runWithWorkerFactory(
}
runners = append(runners, schedulers...)
}
if unhealthy, ok := opened.activity.(activitypool.UnhealthyRemover); ok && !nilInterface(unhealthy) {
reaper, reaperErr := controllerHealth.NewConfiguredUnhealthyReaper(
configurationStore,
unhealthy,
controllerHealth.UnhealthyReaperOptions{
PollInterval: checkSchedulerPollInterval, BatchSize: checkSchedulerBatchSize, Now: options.Now,
},
)
if reaperErr != nil {
return fmt.Errorf("%w: build unhealthy reaper: %w", ErrStartup, reaperErr)
}
runners = append(runners, reaper)
}
}
runners = append(runners, supervisor)
group, err := lifecycle.NewGroup(runners...)

86
internal/controller/health/unhealthy_reaper.go Normal file
View File

@ -0,0 +1,86 @@
package health
import (
"context"
"errors"
"time"
"proxy-pool/internal/config"
"proxy-pool/internal/domain/activitypool"
)
var ErrInvalidUnhealthyReaper = errors.New("invalid unhealthy reaper")
type UnhealthyReaperOptions struct {
PollInterval time.Duration
BatchSize int
Now func() time.Time
}
// ConfiguredUnhealthyReaper applies the current upstream grace policies to a
// bounded activity-pool sweep. It performs no work when every policy disables
// removal, making the default behavior recovery-only.
type ConfiguredUnhealthyReaper struct {
configuration ConfigurationSource
store activitypool.UnhealthyRemover
options UnhealthyReaperOptions
}
func NewConfiguredUnhealthyReaper(
configuration ConfigurationSource,
store activitypool.UnhealthyRemover,
options UnhealthyReaperOptions,
) (*ConfiguredUnhealthyReaper, error) {
if nilInterface(configuration) || nilInterface(store) || options.PollInterval <= 0 || options.BatchSize <= 0 || options.Now == nil {
return nil, ErrInvalidUnhealthyReaper
}
return &ConfiguredUnhealthyReaper{configuration: configuration, store: store, options: options}, nil
}
func (reaper *ConfiguredUnhealthyReaper) Tick(ctx context.Context) (activitypool.UnhealthySweepResult, error) {
if ctx == nil || reaper == nil || nilInterface(reaper.configuration) || nilInterface(reaper.store) ||
reaper.options.PollInterval <= 0 || reaper.options.BatchSize <= 0 || reaper.options.Now == nil {
return activitypool.UnhealthySweepResult{}, ErrInvalidUnhealthyReaper
}
if err := ctx.Err(); err != nil {
return activitypool.UnhealthySweepResult{}, err
}
now := reaper.options.Now()
if now.IsZero() {
return activitypool.UnhealthySweepResult{}, ErrInvalidUnhealthyReaper
}
policies := unhealthyRemovalPolicies(reaper.configuration.Current())
if len(policies) == 0 {
return activitypool.UnhealthySweepResult{}, nil
}
return reaper.store.SweepUnhealthy(ctx, activitypool.UnhealthySweepCommand{
Now: now.UTC(), Limit: reaper.options.BatchSize, RemoveAfterByUpstream: policies,
})
}
func (reaper *ConfiguredUnhealthyReaper) Run(ctx context.Context) error {
if reaper == nil {
return ErrInvalidUnhealthyReaper
}
return runScheduler(ctx, reaper.options.PollInterval, func(tickCtx context.Context) (TickResult, error) {
_, err := reaper.Tick(tickCtx)
return TickResult{}, err
})
}
func unhealthyRemovalPolicies(configuration *config.Config) map[string]time.Duration {
if configuration == nil {
return nil
}
policies := make(map[string]time.Duration)
for upstreamID, upstream := range configuration.Upstreams {
if !upstream.Enabled {
continue
}
check := config.EffectiveCheck(configuration.Defaults.Check, upstream.Check)
if removeAfter := check.UnhealthyRemoveAfter.Value(); removeAfter > 0 {
policies[upstreamID] = removeAfter
}
}
return policies
}

64
internal/controller/health/unhealthy_reaper_test.go Normal file
View File

@ -0,0 +1,64 @@
package health
import (
"context"
"testing"
"time"
"proxy-pool/internal/config"
"proxy-pool/internal/domain/activitypool"
)
type unhealthyRemoverStub struct {
command activitypool.UnhealthySweepCommand
calls int
result activitypool.UnhealthySweepResult
}
func (store *unhealthyRemoverStub) SweepUnhealthy(
_ context.Context,
command activitypool.UnhealthySweepCommand,
) (activitypool.UnhealthySweepResult, error) {
store.calls++
store.command = command
return store.result, nil
}
func TestConfiguredUnhealthyReaperUsesEffectiveEnabledPolicies(t *testing.T) {
now := time.Date(2026, 8, 2, 14, 0, 0, 0, time.UTC)
store := &unhealthyRemoverStub{result: activitypool.UnhealthySweepResult{Removed: 2}}
reaper, err := NewConfiguredUnhealthyReaper(&configurationSourceStub{configuration: &config.Config{
Defaults: config.Defaults{Check: config.Check{UnhealthyRemoveAfter: config.Duration(2 * time.Minute)}},
Upstreams: map[string]config.Upstream{
"provider-a": {Enabled: true, Check: config.Check{UnhealthyRemoveAfter: config.Duration(3 * time.Minute)}},
"provider-b": {Enabled: true},
"provider-c": {Enabled: false},
},
}}, store, UnhealthyReaperOptions{PollInterval: time.Second, BatchSize: 32, Now: func() time.Time { return now }})
if err != nil {
t.Fatalf("NewConfiguredUnhealthyReaper() = %v", err)
}
result, err := reaper.Tick(context.Background())
if err != nil || result != (activitypool.UnhealthySweepResult{Removed: 2}) {
t.Fatalf("Tick() = %+v, %v", result, err)
}
if store.calls != 1 || !store.command.Now.Equal(now) || store.command.Limit != 32 ||
store.command.RemoveAfterByUpstream["provider-a"] != 3*time.Minute ||
store.command.RemoveAfterByUpstream["provider-b"] != 2*time.Minute ||
len(store.command.RemoveAfterByUpstream) != 2 {
t.Fatalf("SweepUnhealthy() command = %+v", store.command)
}
}
func TestConfiguredUnhealthyReaperSkipsDisabledRemoval(t *testing.T) {
store := &unhealthyRemoverStub{}
reaper, err := NewConfiguredUnhealthyReaper(&configurationSourceStub{configuration: &config.Config{
Upstreams: map[string]config.Upstream{"provider-a": {Enabled: true}},
}}, store, UnhealthyReaperOptions{PollInterval: time.Second, BatchSize: 1, Now: time.Now})
if err != nil {
t.Fatalf("NewConfiguredUnhealthyReaper() = %v", err)
}
if result, err := reaper.Tick(context.Background()); err != nil || result != (activitypool.UnhealthySweepResult{}) || store.calls != 0 {
t.Fatalf("Tick(disabled) = (%+v, %v), calls=%d", result, err, store.calls)
}
}

56
internal/domain/activitypool/global_health_test.go
View File

@ -21,7 +21,8 @@ func TestMemoryPoolAppliesGlobalObservationAtomically(t *testing.T) {
failed, err := pool.ApplyGlobalObservation(context.Background(), GlobalHealthCommand{
Observation: healthObservation("task-1", false, now.Add(2*time.Second)), MaxConsecutiveFailures: 3,
})
if err != nil || failed.State != proxyDomain.StateUnhealthy || failed.GlobalHealth.ConsecutiveFailures != 1 {
if err != nil || failed.State != proxyDomain.StateUnhealthy || failed.GlobalHealth.ConsecutiveFailures != 1 ||
!failed.GlobalHealth.UnhealthySince.Equal(now.Add(2*time.Second)) {
t.Fatalf("ApplyGlobalObservation(initial failure) = %+v, %v", failed, err)
}
if _, err := pool.ApplyHealth(context.Background(), HealthUpdate{
@ -33,11 +34,64 @@ func TestMemoryPoolAppliesGlobalObservationAtomically(t *testing.T) {
Observation: healthObservation("task-2", true, now.Add(4*time.Second)), MaxConsecutiveFailures: 3,
})
if err != nil || available.State != proxyDomain.StateAvailable || available.GlobalHealth.ConsecutiveFailures != 0 ||
!available.GlobalHealth.UnhealthySince.IsZero() ||
available.Proxy.LastSuccessAt == nil || !available.Proxy.LastSuccessAt.Equal(now.Add(4*time.Second)) {
t.Fatalf("ApplyGlobalObservation(recovery) = %+v, %v", available, err)
}
}
func TestMemoryPoolSweepsOnlyUnownedProxiesPastConfiguredUnhealthyGrace(t *testing.T) {
now := time.Date(2026, 8, 2, 13, 0, 0, 0, time.UTC)
pool := seededHealthPool(t, now)
if _, err := pool.ApplyHealth(context.Background(), HealthUpdate{
ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking,
}); err != nil {
t.Fatalf("ApplyHealth(checking) = %v", err)
}
if _, err := pool.ApplyGlobalObservation(context.Background(), GlobalHealthCommand{
Observation: healthObservation("task-available", true, now.Add(2*time.Second)), MaxConsecutiveFailures: 1,
}); err != nil {
t.Fatalf("ApplyGlobalObservation(available) = %v", err)
}
assignment, err := pool.Assign(context.Background(), now.Add(3*time.Second), "proxy-a", "worker-a", time.Minute)
if err != nil {
t.Fatalf("Assign() = %v", err)
}
if _, err := pool.ApplyGlobalObservation(context.Background(), GlobalHealthCommand{
Observation: healthObservation("task-suspect", false, now.Add(4*time.Second)), MaxConsecutiveFailures: 2,
}); err != nil {
t.Fatalf("ApplyGlobalObservation(suspect) = %v", err)
}
if _, err := pool.ApplyGlobalObservation(context.Background(), GlobalHealthCommand{
Observation: healthObservation("task-unhealthy", false, now.Add(5*time.Second)), MaxConsecutiveFailures: 2,
}); err != nil {
t.Fatalf("ApplyGlobalObservation(unhealthy) = %v", err)
}
command := UnhealthySweepCommand{
Now: now.Add(7 * time.Second), Limit: 8, RemoveAfterByUpstream: map[string]time.Duration{"provider-a": time.Second},
}
if result, err := pool.SweepUnhealthy(context.Background(), command); err != nil || result != (UnhealthySweepResult{DeferredOwned: 1}) {
t.Fatalf("SweepUnhealthy(owned) = %+v, %v", result, err)
}
if _, exists := pool.entryByIDLocked("proxy-a"); !exists {
t.Fatal("owned unhealthy proxy was removed")
}
if _, err := pool.BeginDrain(context.Background(), "proxy-a", "worker-a", assignment.Epoch); err != nil {
t.Fatalf("BeginDrain() = %v", err)
}
if err := pool.AcknowledgeDrain(context.Background(), "proxy-a", "worker-a", assignment.Epoch, 0, 0); err != nil {
t.Fatalf("AcknowledgeDrain() = %v", err)
}
command.Now = now.Add(9 * time.Second)
if result, err := pool.SweepUnhealthy(context.Background(), command); err != nil || result != (UnhealthySweepResult{Removed: 1}) {
t.Fatalf("SweepUnhealthy(unowned) = %+v, %v", result, err)
}
if _, exists := pool.entryByIDLocked("proxy-a"); exists {
t.Fatal("unowned unhealthy proxy remains after grace")
}
}
func TestMemoryPoolGlobalObservationReplayDoesNotIncreaseFailureStreak(t *testing.T) {
now := time.Date(2026, 7, 31, 11, 0, 0, 0, time.UTC)
pool := seededHealthPool(t, now)

126
internal/domain/activitypool/pool.go
View File

@ -121,6 +121,29 @@ type Maintainer interface {
SweepExpired(context.Context, time.Time, int) (int, error)
}
// UnhealthySweepCommand requests a bounded Controller-side sweep of proxies
// that have remained globally unhealthy past each upstream's configured grace
// period. A policy is absent when the upstream has removal disabled.
type UnhealthySweepCommand struct {
Now time.Time
Limit int
RemoveAfterByUpstream map[string]time.Duration
}
// UnhealthySweepResult separates completed removals from candidates still
// owned by a Worker. Owned entries are intentionally deferred so a stale
// health result never tears down an active Gateway proxy.
type UnhealthySweepResult struct {
Removed int
DeferredOwned int
}
type UnhealthyRemover interface {
SweepUnhealthy(context.Context, UnhealthySweepCommand) (UnhealthySweepResult, error)
}
const maximumUnhealthySweepScan = 1024
type Entry struct {
Proxy proxyDomain.Proxy
UsableUntil time.Time
@ -137,6 +160,7 @@ type MemoryPool struct {
idempotent map[string]idempotencyEntry
ownership map[string]ownershipDomain.Assignment
targets map[targetHealthKey]healthDomain.TargetState
unhealthy map[string]time.Time
nextEpoch uint64
}
@ -160,6 +184,7 @@ var (
_ InventoryReader = (*MemoryPool)(nil)
_ StateInventoryReader = (*MemoryPool)(nil)
_ Maintainer = (*MemoryPool)(nil)
_ UnhealthyRemover = (*MemoryPool)(nil)
_ extractionDomain.Store = (*MemoryPool)(nil)
_ ownershipDomain.Repository = (*MemoryPool)(nil)
)
@ -171,6 +196,7 @@ func NewMemoryPool() *MemoryPool {
idempotent: make(map[string]idempotencyEntry),
ownership: make(map[string]ownershipDomain.Assignment),
targets: make(map[targetHealthKey]healthDomain.TargetState),
unhealthy: make(map[string]time.Time),
}
}
@ -395,6 +421,11 @@ func (p *MemoryPool) ApplyGlobalObservation(
entry.Proxy.LastSuccessAt = timePointer(normalized.ObservedAt)
}
entry.GlobalHealth = next
if next.State == proxyDomain.StateUnhealthy && !next.UnhealthySince.IsZero() {
p.unhealthy[normalized.ProxyID] = next.UnhealthySince
} else {
delete(p.unhealthy, normalized.ProxyID)
}
p.setEntryByIDLocked(normalized.ProxyID, entry)
entry.Proxy = cloneProxy(entry.Proxy)
return entry, nil
@ -564,6 +595,88 @@ func (p *MemoryPool) SweepExpired(ctx context.Context, now time.Time, limit int)
return len(expiredIDs), nil
}
// SweepUnhealthy removes only unowned proxies which remain unhealthy beyond
// the configured upstream grace period. It is bounded by Limit and never
// reaches the Gateway request path.
func (p *MemoryPool) SweepUnhealthy(
ctx context.Context,
command UnhealthySweepCommand,
) (UnhealthySweepResult, error) {
if ctx == nil {
return UnhealthySweepResult{}, ErrInvalidMaintenance
}
if err := ctx.Err(); err != nil {
return UnhealthySweepResult{}, err
}
if p == nil || command.Now.IsZero() || command.Limit <= 0 || !validUnhealthySweepPolicies(command.RemoveAfterByUpstream) {
return UnhealthySweepResult{}, ErrInvalidMaintenance
}
p.mu.Lock()
defer p.mu.Unlock()
if err := ctx.Err(); err != nil {
return UnhealthySweepResult{}, err
}
p.purgeExpiredLocked(command.Now)
type candidate struct {
proxyID string
dueAt time.Time
}
scanLimit := command.Limit
if scanLimit > maximumUnhealthySweepScan/4 {
scanLimit = maximumUnhealthySweepScan
} else {
scanLimit *= 4
}
candidates := make([]candidate, 0, min(len(p.unhealthy), scanLimit))
scanned := 0
for proxyID, dueAt := range p.unhealthy {
if scanned >= scanLimit {
break
}
scanned++
entry, exists := p.entryByIDLocked(proxyID)
if !exists || entry.State != proxyDomain.StateUnhealthy {
delete(p.unhealthy, proxyID)
continue
}
removeAfter, configured := command.RemoveAfterByUpstream[entry.Proxy.SourceUpstream]
if !configured || entry.GlobalHealth.UnhealthySince.IsZero() {
delete(p.unhealthy, proxyID)
continue
}
if dueAt.After(command.Now) || entry.GlobalHealth.UnhealthySince.Add(removeAfter).After(command.Now) {
continue
}
candidates = append(candidates, candidate{proxyID: proxyID, dueAt: dueAt})
}
sort.Slice(candidates, func(left, right int) bool {
if candidates[left].dueAt.Equal(candidates[right].dueAt) {
return candidates[left].proxyID < candidates[right].proxyID
}
return candidates[left].dueAt.Before(candidates[right].dueAt)
})
if len(candidates) > command.Limit {
candidates = candidates[:command.Limit]
}
result := UnhealthySweepResult{}
for _, candidate := range candidates {
entry, exists := p.entryByIDLocked(candidate.proxyID)
if !exists || entry.State != proxyDomain.StateUnhealthy {
delete(p.unhealthy, candidate.proxyID)
continue
}
if entry.OwnerWorkerID != "" {
// Retry later without losing the authoritative first-unhealthy timestamp.
p.unhealthy[candidate.proxyID] = command.Now.UTC().Add(time.Second)
result.DeferredOwned++
continue
}
p.removeEntryByIDLocked(candidate.proxyID)
result.Removed++
}
return result, nil
}
func (p *MemoryPool) Snapshot(now time.Time) []Entry {
if p == nil {
return nil
@ -899,6 +1012,7 @@ func (p *MemoryPool) removeEntryByIDLocked(proxyID string) {
func (p *MemoryPool) removeEntryLocked(key string, entry Entry) {
delete(p.ownership, entry.Proxy.ID)
delete(p.unhealthy, entry.Proxy.ID)
for target := range p.targets {
if target.proxyID == entry.Proxy.ID {
delete(p.targets, target)
@ -908,6 +1022,18 @@ func (p *MemoryPool) removeEntryLocked(key string, entry Entry) {
delete(p.entries, key)
}
func validUnhealthySweepPolicies(policies map[string]time.Duration) bool {
if len(policies) == 0 {
return false
}
for upstreamID, removeAfter := range policies {
if upstreamID == "" || removeAfter <= 0 {
return false
}
}
return true
}
func (p *MemoryPool) entryByIDLocked(proxyID string) (Entry, bool) {
key, ok := p.keyByID[proxyID]
if !ok {

11
internal/domain/health/health.go
View File

@ -75,6 +75,7 @@ type TargetProfile struct {
type GlobalState struct {
State proxyDomain.State
ConsecutiveFailures int
UnhealthySince time.Time
LastTaskID string
LastObservedAt time.Time
LastObservationDigest [sha256.Size]byte
@ -230,6 +231,7 @@ func ReduceGlobal(current GlobalState, observation Observation, maxConsecutiveFa
case proxyDomain.StateChecking, proxyDomain.StateAvailable, proxyDomain.StateSuspect:
next.State = proxyDomain.StateAvailable
next.ConsecutiveFailures = 0
next.UnhealthySince = time.Time{}
return next, nil
default:
return GlobalState{}, ErrInvalidGlobalState
@ -239,11 +241,17 @@ func ReduceGlobal(current GlobalState, observation Observation, maxConsecutiveFa
case proxyDomain.StateChecking:
next.ConsecutiveFailures++
next.State = proxyDomain.StateUnhealthy
if next.UnhealthySince.IsZero() {
next.UnhealthySince = normalized.ObservedAt
}
return next, nil
case proxyDomain.StateAvailable, proxyDomain.StateSuspect:
next.ConsecutiveFailures++
if next.ConsecutiveFailures >= maxConsecutiveFailures {
next.State = proxyDomain.StateUnhealthy
if next.UnhealthySince.IsZero() {
next.UnhealthySince = normalized.ObservedAt
}
} else {
next.State = proxyDomain.StateSuspect
}
@ -330,7 +338,8 @@ func checkObservationOrder(
}
func validateGlobalState(value GlobalState) error {
if value.ConsecutiveFailures < 0 {
if value.ConsecutiveFailures < 0 ||
(!value.LastObservedAt.IsZero() && value.UnhealthySince.After(value.LastObservedAt)) {
return ErrInvalidGlobalState
}
switch value.State {

8
internal/domain/health/health_test.go
View File

@ -16,7 +16,8 @@ func TestReduceGlobalAppliesSuspectUnhealthyAndRecovery(t *testing.T) {
t.Fatalf("first failure = %+v, %v; want SUSPECT / 1", first, err)
}
second, err := ReduceGlobal(first, globalObservation("task-2", false, now.Add(time.Second)), 2)
if err != nil || second.State != proxyDomain.StateUnhealthy || second.ConsecutiveFailures != 2 {
if err != nil || second.State != proxyDomain.StateUnhealthy || second.ConsecutiveFailures != 2 ||
!second.UnhealthySince.Equal(now.Add(time.Second)) {
t.Fatalf("second failure = %+v, %v; want UNHEALTHY / 2", second, err)
}
if next, err := BeginGlobalCheck(second); err != nil || next != proxyDomain.StateChecking {
@ -25,7 +26,8 @@ func TestReduceGlobalAppliesSuspectUnhealthyAndRecovery(t *testing.T) {
recovering := second
recovering.State = proxyDomain.StateChecking
recovered, err := ReduceGlobal(recovering, globalObservation("task-3", true, now.Add(2*time.Second)), 2)
if err != nil || recovered.State != proxyDomain.StateAvailable || recovered.ConsecutiveFailures != 0 {
if err != nil || recovered.State != proxyDomain.StateAvailable || recovered.ConsecutiveFailures != 0 ||
!recovered.UnhealthySince.IsZero() {
t.Fatalf("recovery = %+v, %v; want AVAILABLE / 0", recovered, err)
}
}
@ -33,7 +35,7 @@ func TestReduceGlobalAppliesSuspectUnhealthyAndRecovery(t *testing.T) {
func TestReduceGlobalMarksFailedInitialCheckUnhealthy(t *testing.T) {
now := time.Date(2026, 7, 31, 10, 0, 0, 0, time.UTC)
next, err := ReduceGlobal(GlobalState{State: proxyDomain.StateChecking}, globalObservation("task-1", false, now), 3)
if err != nil || next.State != proxyDomain.StateUnhealthy || next.ConsecutiveFailures != 1 {
if err != nil || next.State != proxyDomain.StateUnhealthy || next.ConsecutiveFailures != 1 || !next.UnhealthySince.Equal(now) {
t.Fatalf("initial failure = %+v, %v; want UNHEALTHY / 1", next, err)
}
}

16
progress.md
View File

@ -1,5 +1,21 @@
# 项目进度
## 2026-08-02
- 已提交并推送 `2af504f feat: expose gateway outcome metrics`：Gateway 请求路径使用
领域 Metrics Observer，将固定阶段的成功/失败聚合到
`proxy_pool_gateway_outcomes_total{stage,result}`，并暴露本地 Outcome 队列满时的
`proxy_pool_gateway_outcome_queue_dropped_total`。指标不包含 Proxy、路由、目标、客户端或凭据。
- 新增队列、Prometheus 注册复用和 Gateway 启动装配测试；`go test ./...` 与
`scripts/verify.ps1` 全部通过，后者按环境跳过 `CGO_ENABLED` 未启用的 race 测试。
- 已完成持续 `UNHEALTHY` 后的安全淘汰：首次异常时间写入 Redis TTL 活动池，
`check.unhealthyRemoveAfter` 按 Upstream 生效；Controller 通过公用、有界的
Reaper 仅回收未分配代理。仍持有 Worker ownership 的候选会延后，待既有
Drain/ACK 清空 ownership 后再删除，避免影响仍在 Gateway Snapshot 中使用的代理。
- Redis 集成标签测试在未设置 `PROXY_POOL_TEST_REDIS_URL` 时现能正确跳过所有契约
子测试；`go test ./...`、`go test -tags=integration ./internal/adapters/redisactivity`
与 `scripts/verify.ps1` 已通过，race 仍由 Linux CI 执行。
## 2026-07-30
- Gateway `Capacity` 新增一次打包原子读取，`snapshot.Store` 可生成完整稀疏

5
task_plan.md
View File

@ -37,6 +37,11 @@
Redis 活动池、低基数状态聚合、Distribution/Admin/Metrics 启动与关闭已完成，
双存储 bootstrap 和探针集成已通过；Worker 运行态存储与权威容量读取原语已
完成，WorkerControlPlane、Provider、业务指标与完整容器进程链仍待实现
14. [已完成] 收敛健康故障生命周期：将 `UNHEALTHY` 的持续时长纳入权威活动池状态，
通过 Controller 后台的有界回收清理无 Worker ownership 的持续异常 Proxy；拥有
Worker 的候选延后到既有 Drain/ACK 清除所有权后再处理。
15. [进行中] 收敛 Worker 发布生命周期：补齐权威 Proxy/Routing 的增量发布和
自动 Drain 编排，使配置停用、健康淘汰与 Snapshot 撤销形成可观测闭环。
## 串并行关系