From 84ed10bd7a1ebc7e7b518b55c6253456bed9bb50 Mon Sep 17 00:00:00 2001 From: youfak Date: Sun, 2 Aug 2026 10:15:25 +0800 Subject: [PATCH] feat: reap sustained unhealthy proxies --- README.md | 4 +- docs/configuration/reference.md | 8 ++ docs/development/implementation-plan.md | 8 +- docs/operations/runbook.md | 11 ++ docs/requirements/traceability.md | 2 +- findings.md | 14 ++ internal/adapters/redisactivity/codec.go | 5 +- internal/adapters/redisactivity/extract.go | 2 +- .../global_health_integration_test.go | 97 +++++++++++++- internal/adapters/redisactivity/health.go | 5 +- internal/adapters/redisactivity/keys.go | 2 + .../adapters/redisactivity/maintenance.go | 58 +++++++- internal/adapters/redisactivity/ownership.go | 2 +- .../runtime_contract_integration_test.go | 8 +- internal/adapters/redisactivity/scripts.go | 5 +- .../redisactivity/scripts/extract.lua | 14 +- .../adapters/redisactivity/scripts/health.lua | 13 ++ .../redisactivity/scripts/ownership.lua | 2 + .../adapters/redisactivity/scripts/status.lua | 2 + .../adapters/redisactivity/scripts/sweep.lua | 44 ++++++ .../adapters/redisactivity/scripts/upsert.lua | 3 + internal/adapters/redisactivity/status.go | 2 +- internal/adapters/redisactivity/upsert.go | 4 + internal/config/config.go | 1 + internal/config/config_test.go | 7 + internal/config/effective_check.go | 3 + internal/config/effective_check_test.go | 8 +- internal/config/validate.go | 4 + internal/controller/bootstrap/bootstrap.go | 13 ++ .../controller/health/unhealthy_reaper.go | 86 ++++++++++++ .../health/unhealthy_reaper_test.go | 64 +++++++++ .../domain/activitypool/global_health_test.go | 56 +++++++- internal/domain/activitypool/pool.go | 126 ++++++++++++++++++ internal/domain/health/health.go | 11 +- internal/domain/health/health_test.go | 8 +- progress.md | 16 +++ task_plan.md | 5 + 37 files changed, 689 insertions(+), 34 deletions(-) create mode 100644 internal/controller/health/unhealthy_reaper.go create mode 100644 internal/controller/health/unhealthy_reaper_test.go diff --git a/README.md b/README.md index d78616d..336f221 100644 --- a/README.md +++ b/README.md @@ -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 进程拓扑， diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index fdf642d..168cca5 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -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 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index d402b92..21ecadb 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -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 后已启用上游的策略变更、停用，以及新启用上游都无需重启 diff --git a/docs/operations/runbook.md b/docs/operations/runbook.md index 99ae977..c3067f4 100644 --- a/docs/operations/runbook.md +++ b/docs/operations/runbook.md @@ -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 管理 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index 999af15..e6af7e6 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -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 测试 | diff --git a/findings.md b/findings.md index d72c2a2..d7f5865 100644 --- a/findings.md +++ b/findings.md @@ -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`。 diff --git a/internal/adapters/redisactivity/codec.go b/internal/adapters/redisactivity/codec.go index 78408dc..875c97e 100644 --- a/internal/adapters/redisactivity/codec.go +++ b/internal/adapters/redisactivity/codec.go @@ -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 diff --git a/internal/adapters/redisactivity/extract.go b/internal/adapters/redisactivity/extract.go index e0601f2..d567713 100644 --- a/internal/adapters/redisactivity/extract.go +++ b/internal/adapters/redisactivity/extract.go @@ -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 diff --git a/internal/adapters/redisactivity/global_health_integration_test.go b/internal/adapters/redisactivity/global_health_integration_test.go index a119e15..b75ff63 100644 --- a/internal/adapters/redisactivity/global_health_integration_test.go +++ b/internal/adapters/redisactivity/global_health_integration_test.go @@ -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, diff --git a/internal/adapters/redisactivity/health.go b/internal/adapters/redisactivity/health.go index 505b3bc..e567503 100644 --- a/internal/adapters/redisactivity/health.go +++ b/internal/adapters/redisactivity/health.go @@ -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[:])) diff --git a/internal/adapters/redisactivity/keys.go b/internal/adapters/redisactivity/keys.go index 93a9481..54d4978 100644 --- a/internal/adapters/redisactivity/keys.go +++ b/internal/adapters/redisactivity/keys.go @@ -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", diff --git a/internal/adapters/redisactivity/maintenance.go b/internal/adapters/redisactivity/maintenance.go index bc76160..257f58f 100644 --- a/internal/adapters/redisactivity/maintenance.go +++ b/internal/adapters/redisactivity/maintenance.go @@ -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 } diff --git a/internal/adapters/redisactivity/ownership.go b/internal/adapters/redisactivity/ownership.go index a985a38..96e0166 100644 --- a/internal/adapters/redisactivity/ownership.go +++ b/internal/adapters/redisactivity/ownership.go @@ -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 { diff --git a/internal/adapters/redisactivity/runtime_contract_integration_test.go b/internal/adapters/redisactivity/runtime_contract_integration_test.go index 3a02692..4f5e55e 100644 --- a/internal/adapters/redisactivity/runtime_contract_integration_test.go +++ b/internal/adapters/redisactivity/runtime_contract_integration_test.go @@ -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, diff --git a/internal/adapters/redisactivity/scripts.go b/internal/adapters/redisactivity/scripts.go index c8f0006..aabd08a 100644 --- a/internal/adapters/redisactivity/scripts.go +++ b/internal/adapters/redisactivity/scripts.go @@ -88,8 +88,9 @@ type ownershipScriptReply struct { } type maintenanceScriptReply struct { - Status scriptStatus `json:"status"` - Count int `json:"count"` + Status scriptStatus `json:"status"` + Count int `json:"count"` + DeferredOwned int `json:"deferredOwned"` } type statusScriptReply struct { diff --git a/internal/adapters/redisactivity/scripts/extract.lua b/internal/adapters/redisactivity/scripts/extract.lua index f60eec0..c830eea 100644 --- a/internal/adapters/redisactivity/scripts/extract.lua +++ b/internal/adapters/redisactivity/scripts/extract.lua @@ -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 diff --git a/internal/adapters/redisactivity/scripts/health.lua b/internal/adapters/redisactivity/scripts/health.lua index 50ba872..9ea8a9e 100644 --- a/internal/adapters/redisactivity/scripts/health.lua +++ b/internal/adapters/redisactivity/scripts/health.lua @@ -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 diff --git a/internal/adapters/redisactivity/scripts/ownership.lua b/internal/adapters/redisactivity/scripts/ownership.lua index bdeb4e1..ebf3487 100644 --- a/internal/adapters/redisactivity/scripts/ownership.lua +++ b/internal/adapters/redisactivity/scripts/ownership.lua @@ -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 diff --git a/internal/adapters/redisactivity/scripts/status.lua b/internal/adapters/redisactivity/scripts/status.lua index 8c668e6..cb056a8 100644 --- a/internal/adapters/redisactivity/scripts/status.lua +++ b/internal/adapters/redisactivity/scripts/status.lua @@ -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 diff --git a/internal/adapters/redisactivity/scripts/sweep.lua b/internal/adapters/redisactivity/scripts/sweep.lua index 01417f8..5cac2d2 100644 --- a/internal/adapters/redisactivity/scripts/sweep.lua +++ b/internal/adapters/redisactivity/scripts/sweep.lua @@ -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 diff --git a/internal/adapters/redisactivity/scripts/upsert.lua b/internal/adapters/redisactivity/scripts/upsert.lua index bb83cbf..dd8fe32 100644 --- a/internal/adapters/redisactivity/scripts/upsert.lua +++ b/internal/adapters/redisactivity/scripts/upsert.lua @@ -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 diff --git a/internal/adapters/redisactivity/status.go b/internal/adapters/redisactivity/status.go index 31b95e6..a2b9b01 100644 --- a/internal/adapters/redisactivity/status.go +++ b/internal/adapters/redisactivity/status.go @@ -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 diff --git a/internal/adapters/redisactivity/upsert.go b/internal/adapters/redisactivity/upsert.go index e76e4b5..d09fe6b 100644 --- a/internal/adapters/redisactivity/upsert.go +++ b/internal/adapters/redisactivity/upsert.go @@ -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 { diff --git a/internal/config/config.go b/internal/config/config.go index 6d0c4b6..9dc2c13 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -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"` } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 569e93e..1cb2114 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -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 { diff --git a/internal/config/effective_check.go b/internal/config/effective_check.go index b732051..27ab002 100644 --- a/internal/config/effective_check.go +++ b/internal/config/effective_check.go @@ -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 { diff --git a/internal/config/effective_check_test.go b/internal/config/effective_check_test.go index 0d3af33..c39faa9 100644 --- a/internal/config/effective_check_test.go +++ b/internal/config/effective_check_test.go @@ -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) { diff --git a/internal/config/validate.go b/internal/config/validate.go index 9fec7af..299c4b5 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -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 } diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index 4d58aa9..d56da3e 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -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...) diff --git a/internal/controller/health/unhealthy_reaper.go b/internal/controller/health/unhealthy_reaper.go new file mode 100644 index 0000000..d09ce5a --- /dev/null +++ b/internal/controller/health/unhealthy_reaper.go @@ -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 +} diff --git a/internal/controller/health/unhealthy_reaper_test.go b/internal/controller/health/unhealthy_reaper_test.go new file mode 100644 index 0000000..744bf17 --- /dev/null +++ b/internal/controller/health/unhealthy_reaper_test.go @@ -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) + } +} diff --git a/internal/domain/activitypool/global_health_test.go b/internal/domain/activitypool/global_health_test.go index aaa7972..cc2f854 100644 --- a/internal/domain/activitypool/global_health_test.go +++ b/internal/domain/activitypool/global_health_test.go @@ -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) diff --git a/internal/domain/activitypool/pool.go b/internal/domain/activitypool/pool.go index 38132b4..832fcea 100644 --- a/internal/domain/activitypool/pool.go +++ b/internal/domain/activitypool/pool.go @@ -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 { diff --git a/internal/domain/health/health.go b/internal/domain/health/health.go index ee0ac34..3b38c1a 100644 --- a/internal/domain/health/health.go +++ b/internal/domain/health/health.go @@ -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 { diff --git a/internal/domain/health/health_test.go b/internal/domain/health/health_test.go index 837dafe..891deed 100644 --- a/internal/domain/health/health_test.go +++ b/internal/domain/health/health_test.go @@ -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) } } diff --git a/progress.md b/progress.md index 206fcf3..6f320dc 100644 --- a/progress.md +++ b/progress.md @@ -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` 可生成完整稀疏 diff --git a/task_plan.md b/task_plan.md index c3864a1..09304c9 100644 --- a/task_plan.md +++ b/task_plan.md @@ -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 撤销形成可观测闭环。 ## 串并行关系