From 5ad1d10957c11ba28a91a7c709ee45714c9d8fa3 Mon Sep 17 00:00:00 2001 From: youfak Date: Sun, 2 Aug 2026 12:38:52 +0800 Subject: [PATCH] feat: drain unhealthy worker assignments automatically --- README.md | 5 +- docs/configuration/reference.md | 8 +- docs/development/implementation-plan.md | 7 +- docs/requirements/traceability.md | 2 +- findings.md | 9 +- .../global_health_integration_test.go | 81 +++++++++++++++-- .../adapters/redisactivity/maintenance.go | 14 ++- internal/adapters/redisactivity/ownership.go | 48 ++++++++-- internal/adapters/redisactivity/scripts.go | 19 +++- .../redisactivity/scripts/ownership.lua | 87 ++++++++++++------ .../adapters/redisactivity/scripts/sweep.lua | 20 +++- .../controller/health/unhealthy_reaper.go | 18 +++- .../health/unhealthy_reaper_test.go | 49 +++++++++- .../domain/activitypool/global_health_test.go | 71 ++++++++++++++- internal/domain/activitypool/pool.go | 91 ++++++++++++++++--- progress.md | 5 +- task_plan.md | 4 +- 17 files changed, 442 insertions(+), 96 deletions(-) diff --git a/README.md b/README.md index 37d988d..736a42e 100644 --- a/README.md +++ b/README.md @@ -260,9 +260,10 @@ go run ./cmd/proxy-loadgen ` ownership 索引，以及 Gateway 快照客户端。 - **P0 - Checker 健康链**：BASIC/EGRESS/TARGET 的共享调度、实际探测、Observation reducer 和 `FETCHED -> AVAILABLE / SUSPECT / UNHEALTHY` 状态链已完成；可配置的持续 UNHEALTHY - 回收只删除未分配 Proxy，已分配项等待 Drain/ACK；TARGET 事实按路由目标 Profile 独立归并。 + 回收会自动为已分配项发起带健康/归属栅栏的 Drain，待 Snapshot ACK 与运行态归零后再删除； + TARGET 事实按路由目标 Profile 独立归并。 - **P1 - Gateway 与 Routing**：Gateway 进程、快照凭据分发、五种 Routing 策略与 - `onUnavailable` 已接入；动态容量调整和 Drain 闭环待完成。 + `onUnavailable` 已接入；动态容量调整和配置停用触发的 Drain 编排待完成。 - **P1 - 可观测与部署**：低基数业务指标、完整 Compose/Kubernetes 进程拓扑， 以及故障转移和恢复演练。 - **P2 - 容量证明**：`proxy-loadgen`、HTTP/CONNECT/Extract 分场景压测，以及 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index a2629c3..f7e1ef5 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -482,9 +482,11 @@ proxyAuth: （默认）表示只保留异常状态，永不因该策略删除。首次进入 UNHEALTHY 的时间会跨 CHECKING 重试保持不变，成功恢复 AVAILABLE 时清除。 - 回收器只运行在 Controller 后台，使用 Redis 有序索引和固定批次，不进入 Gateway - 请求热路径。当前只原子删除没有 Worker ownership 的 Proxy；仍属于 Worker 的候选会 - 延后，等待既有 Drain/ACK 完成后再由下一轮回收，避免删除仍可能出现在 Gateway - Snapshot 中的代理。 + 请求热路径。没有 Worker ownership 的 Proxy 在同一清理边界删除；仍属于 Worker 的候选 + 先在有界回执中携带 worker、assignment epoch 和首次异常时间，再由条件式 Drain 原子 + 复核 `UNHEALTHY` 状态与归属未变化后创建 Ticket。后续必须等待排除快照 ACK 与 Runtime + 的 Active/Reserved 归零，才会释放 ownership 并由下一轮回收，避免删除仍可能出现在 + Gateway Snapshot 中的代理。 ## 9. 存储、Admin 与 Metrics diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 21ecadb..f00b4e5 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -263,7 +263,7 @@ WorkerControlPlane gRPC 接收端、session 签发/心跳、Snapshot ACK 账本 Snapshot 流、Gateway 会话客户端与快照凭据分发已完成；权威 Proxy/Routing 发布、 Outcome 上报已完成为有界队列、批次序列/摘要栅栏和确认重试；健康 BASIC/EGRESS/TARGET 执行链已完成；持续 UNHEALTHY 的未分配 Proxy 可由后台有界回收，拥有 Worker -所有权的项仍等待完整 Drain/ACK 触发，不直接删除，因此 Task 10 尚未全部完成。 +所有权的项会通过健康/归属条件栅栏自动发起 Drain，并在排除快照 ACK 与运行态归零后回收。 ## Task 11: Checker and Health Reducer @@ -293,8 +293,9 @@ EGRESS 的出口身份响应解析支持固定上限的纯文本和常见 JSON I 代理过期均沿用同一原子任务边界。BASIC、EGRESS 和 TARGET 分组轮转并共享 Upstream in-flight 上限，因此本任务已完成。持续 UNHEALTHY 回收使用每 Upstream 的 `check.unhealthyRemoveAfter`：全局 Reducer 保存首次进入 UNHEALTHY 的时间，Redis 以 -有序索引有界扫描；只有无 Worker ownership 的项会沿用统一清理边界删除，已拥有 -Worker 的项延后至 Drain/ACK 解除所有权后重试。 +有序索引有界扫描；没有 Worker ownership 的项沿用统一清理边界删除，已拥有 Worker 的 +项通过包含 assignment epoch 与首次异常时间的候选，由条件式 Drain 原子重校验后进入 +Ticket/排除快照/Runtime 零计数闭环。 补充进度（2026-08-02）：BASIC 调度已改为配置驱动监督器。它每轮读取已发布快照并复用 有界派发逻辑，所以 reload 后已启用上游的策略变更、停用，以及新启用上游都无需重启 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index e6af7e6..f989dde 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 活动池原子提交全局连续失败、精确重放和成功恢复；`check.unhealthyRemoveAfter` 驱动后台有界回收，Redis/Memory 复用相同窄接口并只删除无 Worker ownership 的异常项。已拥有 Worker 的项会延后到 Drain/ACK 清除所有权后重试，自动触发 Drain 的发布闭环仍待完成 | +| HEALTH-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | Controller 公用 Reducer 已通过 Memory/Redis 活动池原子提交全局连续失败、精确重放和成功恢复；`check.unhealthyRemoveAfter` 驱动后台有界回收。已拥有 Worker 的候选带 worker、assignment epoch 与首次异常时间进入有界回执，条件式 Drain 在原子重校验后创建 Ticket；排除快照 ACK 与 Runtime 零计数才释放 ownership，下一轮回收删除异常项 | | 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 4a1e6c6..b879043 100644 --- a/findings.md +++ b/findings.md @@ -108,8 +108,9 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两 生效。Controller Reaper 读取当前配置快照，以固定批次调用公用 `activitypool.UnhealthyRemover`，不进入 Gateway 热路径。 - Redis 使用 `{activity}` 槽内的有序索引扫描到期候选；Memory 参考实现保持相同 - `SweepUnhealthy` 契约。两者只删除无 Worker ownership 的记录，仍归属 Worker 的 - 候选延后一秒后再检查，不自行发起 Drain。 + `SweepUnhealthy` 契约。两者只删除无 Worker ownership 的记录；仍归属 Worker 的 + 候选会延后一秒并在受限回执中返回 assignment epoch 与首次异常时间。Reaper 通过公共 + 条件式 Drain 再次核验记录仍为 `UNHEALTHY`、时间戳和归属均未变化，才创建 Ticket。 - Health、Upsert、Extract、Ownership、状态读取与过期清理脚本都在同一原子删除边界 维护该索引。Memory 单测、Redis 集成契约和配置/Reaper 单测覆盖恢复、阈值、延后与 Drain/ACK 后清理。 @@ -122,8 +123,8 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两 - gRPC Handler 每次成功发送完整 Snapshot 都重置服务端到期计时器；流中刷新失败或中断时 Gateway 使用既有 `SessionSupervisor` 退避重连。Gateway 本地 Store 对从完整 Snapshot 消失且仍有 Active/Reserved 的 Proxy 已按 draining 继续上报。Redis Runtime 替换已在当前 - session ACK、Ticket 屏障和完整零计数同时成立时自动释放 owner；Reaper 和配置停用尚未 - 调用 `BeginDrain`，因此全链路的 Drain 触发编排仍未完成。 + session ACK、Ticket 屏障和完整零计数同时成立时自动释放 owner；Reaper 已对持续 + `UNHEALTHY` 的已归属代理调用条件式 Drain。配置停用尚未接入同一触发编排。 - Snapshot 的 `version` 是同一 Worker 流的连续序列，`ownership_epoch` 是独立且只能前进 的权威栅栏；epoch 变化不重置 version。此前 Gateway 错把 epoch 前进要求为 version=1， 与 Controller 的 `lastAppliedVersion+1` 生成规则冲突，现已用连续版本规则统一。 diff --git a/internal/adapters/redisactivity/global_health_integration_test.go b/internal/adapters/redisactivity/global_health_integration_test.go index b75ff63..8290146 100644 --- a/internal/adapters/redisactivity/global_health_integration_test.go +++ b/internal/adapters/redisactivity/global_health_integration_test.go @@ -91,11 +91,11 @@ func TestRedisSweepsUnownedProxiesAfterUnhealthyGrace(t *testing.T) { 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{}) { + if result, err := fixture.Adapter.SweepUnhealthy(context.Background(), command); err != nil || result.Removed != 0 || result.DeferredOwned != 0 || len(result.DrainCandidates) != 0 { 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}) { + if result, err := fixture.Adapter.SweepUnhealthy(context.Background(), command); err != nil || result.Removed != 1 || result.DeferredOwned != 0 || len(result.DrainCandidates) != 0 { 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) { @@ -106,7 +106,7 @@ func TestRedisSweepsUnownedProxiesAfterUnhealthyGrace(t *testing.T) { } } -func TestRedisUnhealthySweepDefersOwnedProxyUntilDrainAcknowledgement(t *testing.T) { +func TestRedisUnhealthySweepStartsConditionalDrainForOwnedProxy(t *testing.T) { fixture := newRedisTestFixture(t) now := redisTestNow() if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ @@ -142,21 +142,88 @@ func TestRedisUnhealthySweepDefersOwnedProxyUntilDrainAcknowledgement(t *testing 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}) { + result, err := fixture.Adapter.SweepUnhealthy(context.Background(), command) + if err != nil || result.Removed != 0 || result.DeferredOwned != 1 || len(result.DrainCandidates) != 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) + candidate := result.DrainCandidates[0] + if candidate.ProxyID != "proxy-a" || candidate.WorkerID != "worker-a" || candidate.AssignmentEpoch != assignment.Epoch || + !candidate.UnhealthySince.Equal(now.Add(5*time.Second)) { + t.Fatalf("SweepUnhealthy() candidate = %+v", candidate) + } + if started, err := fixture.Adapter.BeginUnhealthyDrain(context.Background(), command.Now, candidate); err != nil || !started { + t.Fatalf("BeginUnhealthyDrain() = %t, %v", started, err) + } + if started, err := fixture.Adapter.BeginUnhealthyDrain(context.Background(), command.Now, candidate); err != nil || started { + t.Fatalf("BeginUnhealthyDrain(idempotent) = %t, %v", started, 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}) { + if result, err := fixture.Adapter.SweepUnhealthy(context.Background(), command); err != nil || result.Removed != 1 || result.DeferredOwned != 0 || len(result.DrainCandidates) != 0 { t.Fatalf("SweepUnhealthy(unowned) = %+v, %v", result, err) } } +func TestRedisRejectsUnhealthyDrainCandidateAfterRecovery(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: 1, RemoveAfterByUpstream: map[string]time.Duration{"provider-a": time.Second}, + } + result, err := fixture.Adapter.SweepUnhealthy(context.Background(), command) + if err != nil || len(result.DrainCandidates) != 1 { + t.Fatalf("SweepUnhealthy() = %+v, %v", result, err) + } + if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: now.Add(8 * time.Second), NextState: proxyDomain.StateChecking, + }); err != nil { + t.Fatalf("ApplyHealth(rechecking) = %v", err) + } + if _, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{ + Observation: redisGlobalObservation("task-recovered", true, now.Add(9*time.Second)), MaxConsecutiveFailures: 2, + }); err != nil { + t.Fatalf("ApplyGlobalObservation(recovered) = %v", err) + } + if started, err := fixture.Adapter.BeginUnhealthyDrain(context.Background(), now.Add(9*time.Second), result.DrainCandidates[0]); err != nil || started { + t.Fatalf("BeginUnhealthyDrain(recovered) = %t, %v", started, err) + } + if current, ok, err := fixture.Adapter.Get(context.Background(), "proxy-a"); err != nil || !ok || current.Epoch != assignment.Epoch || current.Draining { + t.Fatalf("Get(recovered) = %+v, %t, %v", current, ok, 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/maintenance.go b/internal/adapters/redisactivity/maintenance.go index 257f58f..9de1e0d 100644 --- a/internal/adapters/redisactivity/maintenance.go +++ b/internal/adapters/redisactivity/maintenance.go @@ -108,10 +108,20 @@ func (a *Adapter) SweepUnhealthy( return activitypool.UnhealthySweepResult{}, activitypool.ErrInvalidMaintenance } if reply.Status != scriptOK || reply.Count < 0 || reply.Count > limit || reply.DeferredOwned < 0 || - reply.DeferredOwned > maximumUnhealthySweepScan { + reply.DeferredOwned > maximumUnhealthySweepScan || len(reply.DrainCandidates) > limit { return activitypool.UnhealthySweepResult{}, invalidScriptReply("unexpected unhealthy sweep reply") } - return activitypool.UnhealthySweepResult{Removed: reply.Count, DeferredOwned: reply.DeferredOwned}, nil + result := activitypool.UnhealthySweepResult{Removed: reply.Count, DeferredOwned: reply.DeferredOwned} + for _, candidate := range reply.DrainCandidates { + if candidate.ProxyID == "" || candidate.WorkerID == "" || candidate.AssignmentEpoch == 0 || candidate.UnhealthySinceMS <= 0 { + return activitypool.UnhealthySweepResult{}, invalidScriptReply("unhealthy sweep reply contained an invalid drain candidate") + } + result.DrainCandidates = append(result.DrainCandidates, activitypool.UnhealthyDrainCandidate{ + ProxyID: candidate.ProxyID, WorkerID: candidate.WorkerID, AssignmentEpoch: candidate.AssignmentEpoch, + UnhealthySince: time.UnixMilli(candidate.UnhealthySinceMS).UTC(), + }) + } + return result, nil } func (a *Adapter) runMaintenance( diff --git a/internal/adapters/redisactivity/ownership.go b/internal/adapters/redisactivity/ownership.go index 4abf685..66c8a93 100644 --- a/internal/adapters/redisactivity/ownership.go +++ b/internal/adapters/redisactivity/ownership.go @@ -8,20 +8,23 @@ import ( "fmt" "time" + "proxy-pool/internal/domain/activitypool" ownershipDomain "proxy-pool/internal/domain/ownership" ) const ( - ownershipAssign = "assign" - ownershipRenew = "renew" - ownershipBeginDrain = "begin_drain" - ownershipAcknowledgeDrain = "acknowledge_drain" - ownershipGet = "get" - ownershipExpire = "expire" + ownershipAssign = "assign" + ownershipRenew = "renew" + ownershipBeginDrain = "begin_drain" + ownershipBeginUnhealthyDrain = "begin_unhealthy_drain" + ownershipAcknowledgeDrain = "acknowledge_drain" + ownershipGet = "get" + ownershipExpire = "expire" ) var _ ownershipDomain.Repository = (*Adapter)(nil) var _ ownershipDomain.DrainTicketStore = (*Adapter)(nil) +var _ activitypool.UnhealthyDrainStarter = (*Adapter)(nil) func (a *Adapter) Assign( ctx context.Context, @@ -112,6 +115,39 @@ func (a *Adapter) BeginDrain( } } +func (a *Adapter) BeginUnhealthyDrain( + ctx context.Context, + now time.Time, + candidate activitypool.UnhealthyDrainCandidate, +) (bool, error) { + if err := validateOwnershipCall(ctx, a); err != nil { + return false, err + } + if now.IsZero() || candidate.ProxyID == "" || candidate.WorkerID == "" || candidate.AssignmentEpoch == 0 || candidate.UnhealthySince.IsZero() { + return false, activitypool.ErrInvalidMaintenance + } + reply, err := a.runOwnership( + ctx, ownershipBeginUnhealthyDrain, true, now.UTC().UnixMilli(), candidate.ProxyID, candidate.WorkerID, + candidate.AssignmentEpoch, candidate.UnhealthySince.UTC().UnixMilli(), 0, 0, + ) + if err != nil { + return false, err + } + switch reply.Status { + case scriptOK: + if _, err := decodeAssignmentReply(reply); err != nil { + return false, err + } + return reply.Started, nil + case scriptStale, scriptNotFound: + return false, nil + case scriptInvalid: + return false, activitypool.ErrInvalidMaintenance + default: + return false, invalidScriptReply("unexpected unhealthy ownership drain status") + } +} + func (a *Adapter) PendingDrains(ctx context.Context, workerID string, limit int) ([]ownershipDomain.DrainTicket, error) { if err := validateOwnershipCall(ctx, a); err != nil { return nil, err diff --git a/internal/adapters/redisactivity/scripts.go b/internal/adapters/redisactivity/scripts.go index cdb7990..8b5b044 100644 --- a/internal/adapters/redisactivity/scripts.go +++ b/internal/adapters/redisactivity/scripts.go @@ -83,8 +83,9 @@ type extractScriptReply struct { } type ownershipScriptReply struct { - Status scriptStatus `json:"status"` - Record string `json:"record,omitempty"` + Status scriptStatus `json:"status"` + Record string `json:"record,omitempty"` + Started bool `json:"started"` } type drainTicketsScriptReply struct { @@ -93,9 +94,17 @@ type drainTicketsScriptReply struct { } type maintenanceScriptReply struct { - Status scriptStatus `json:"status"` - Count int `json:"count"` - DeferredOwned int `json:"deferredOwned"` + Status scriptStatus `json:"status"` + Count int `json:"count"` + DeferredOwned int `json:"deferredOwned"` + DrainCandidates []maintenanceDrainCandidateScriptRow `json:"drainCandidates"` +} + +type maintenanceDrainCandidateScriptRow struct { + ProxyID string `json:"proxyId"` + WorkerID string `json:"workerId"` + AssignmentEpoch uint64 `json:"assignmentEpoch"` + UnhealthySinceMS int64 `json:"unhealthySinceMs"` } type statusScriptReply struct { diff --git a/internal/adapters/redisactivity/scripts/ownership.lua b/internal/adapters/redisactivity/scripts/ownership.lua index 8391e99..c584700 100644 --- a/internal/adapters/redisactivity/scripts/ownership.lua +++ b/internal/adapters/redisactivity/scripts/ownership.lua @@ -240,6 +240,40 @@ local function clear_owner(id, assignment, at_ms, restore) end end +local function begin_drain(current) + if current.draining then + return cjson.encode(current), false, nil + end + if type(worker_draining_index_key) ~= 'string' or worker_draining_index_key == '' or + not string.find(worker_draining_index_key, '{activity}', 1, true) then + return nil, false, 'invalid' + end + current.draining = true + current.assignmentVersion = tonumber(current.assignmentVersion) + 1 + local drain_epoch = redis.call('INCR', epoch_key) + redis.call('PERSIST', epoch_key) + local encoded = cjson.encode(current) + redis.call('HSET', owners_key, proxy_id, encoded) + local record = decode_table(redis.call('HGET', records_key, proxy_id)) + remove_owned(proxy_id, record) + if type(current.workerIndexKey) == 'string' and current.workerIndexKey ~= '' then + redis.call('ZREM', current.workerIndexKey, proxy_id) + end + local ticket = { + version = 1, + proxyId = proxy_id, + workerId = worker_id, + workerIndexKey = worker_draining_index_key, + assignmentEpoch = current.epoch, + requiredSnapshotEpoch = drain_epoch, + } + redis.call('HSET', drain_tickets_key, proxy_id, cjson.encode(ticket)) + redis.call('ZADD', worker_draining_index_key, current.expiresAtMs, proxy_id) + touch(drain_tickets_key, tonumber(current.expiresAtMs)) + touch(worker_draining_index_key, tonumber(current.expiresAtMs)) + return encoded, true, nil +end + if operation == 'assign' then cleanup_hard_expired(now_ms) local current_raw = redis.call('HGET', owners_key, proxy_id) @@ -336,37 +370,30 @@ if operation == 'begin_drain' then if not valid_assignment(current) or current.workerId ~= worker_id or tonumber(current.epoch) ~= epoch then return finish({status = 'stale'}) end - if not current.draining then - if type(worker_draining_index_key) ~= 'string' or worker_draining_index_key == '' or - not string.find(worker_draining_index_key, '{activity}', 1, true) then - return finish({status = 'invalid'}) - end - current.draining = true - current.assignmentVersion = tonumber(current.assignmentVersion) + 1 - local drain_epoch = redis.call('INCR', epoch_key) - redis.call('PERSIST', epoch_key) - local encoded = cjson.encode(current) - redis.call('HSET', owners_key, proxy_id, encoded) - local record = decode_table(redis.call('HGET', records_key, proxy_id)) - remove_owned(proxy_id, record) - if type(current.workerIndexKey) == 'string' and current.workerIndexKey ~= '' then - redis.call('ZREM', current.workerIndexKey, proxy_id) - end - local ticket = { - version = 1, - proxyId = proxy_id, - workerId = worker_id, - workerIndexKey = worker_draining_index_key, - assignmentEpoch = current.epoch, - requiredSnapshotEpoch = drain_epoch, - } - redis.call('HSET', drain_tickets_key, proxy_id, cjson.encode(ticket)) - redis.call('ZADD', worker_draining_index_key, current.expiresAtMs, proxy_id) - touch(drain_tickets_key, tonumber(current.expiresAtMs)) - touch(worker_draining_index_key, tonumber(current.expiresAtMs)) - return finish({status = 'ok', record = encoded}) + local encoded, _, failure = begin_drain(current) + if failure then + return finish({status = failure}) end - return finish({status = 'ok', record = cjson.encode(current)}) + return finish({status = 'ok', record = encoded}) +end + +if operation == 'begin_unhealthy_drain' then + cleanup_hard_expired(now_ms) + local current = decode_table(redis.call('HGET', owners_key, proxy_id)) + if not valid_assignment(current) or current.workerId ~= worker_id or tonumber(current.epoch) ~= epoch or + tonumber(current.expiresAtMs) <= now_ms then + return finish({status = 'stale'}) + end + local record = decode_table(redis.call('HGET', records_key, proxy_id)) + if not record or record.ownerWorkerId ~= worker_id or record.state ~= 'UNHEALTHY' or + not tonumber(record.unhealthySinceMs) or tonumber(record.unhealthySinceMs) ~= value then + return finish({status = 'stale'}) + end + local encoded, started, failure = begin_drain(current) + if failure then + return finish({status = failure}) + end + return finish({status = 'ok', record = encoded, started = started}) end if operation == 'acknowledge_drain' then diff --git a/internal/adapters/redisactivity/scripts/sweep.lua b/internal/adapters/redisactivity/scripts/sweep.lua index 5cac2d2..fd60aa5 100644 --- a/internal/adapters/redisactivity/scripts/sweep.lua +++ b/internal/adapters/redisactivity/scripts/sweep.lua @@ -166,7 +166,7 @@ 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}) + return finish({status = 'invalid', count = 0, deferredOwned = 0, drainCandidates = {}}) end local scan_limit = limit * 4 if scan_limit > 1024 then @@ -175,6 +175,7 @@ if operation == 'unhealthy' then local candidates = redis.call('ZRANGEBYSCORE', health_unhealthy_key, '-inf', now_ms, 'LIMIT', 0, scan_limit) local removed = 0 local deferred_owned = 0 + local drain_candidates = {} for _, proxy_id in ipairs(candidates) do local raw = redis.call('HGET', records_key, proxy_id) if not raw then @@ -195,6 +196,21 @@ if operation == 'unhealthy' then local retry_at_ms = now_ms + 1000 redis.call('ZADD', health_unhealthy_key, retry_at_ms, proxy_id) deferred_owned = deferred_owned + 1 + local owner_valid = false + local owner = nil + if owner_raw then + owner_valid, owner = pcall(cjson.decode, owner_raw) + end + if owner_valid and type(owner) == 'table' and owner.draining ~= true and + type(owner.workerId) == 'string' and owner.workerId ~= '' and tonumber(owner.epoch) and + tonumber(owner.epoch) > 0 and #drain_candidates < limit then + table.insert(drain_candidates, { + proxyId = proxy_id, + workerId = owner.workerId, + assignmentEpoch = tonumber(owner.epoch), + unhealthySinceMs = unhealthy_since_ms, + }) + end elseif removed < limit then remove_proxy(proxy_id) removed = removed + 1 @@ -202,7 +218,7 @@ if operation == 'unhealthy' then end end end - return finish({status = 'ok', count = removed, deferredOwned = deferred_owned}) + return finish({status = 'ok', count = removed, deferredOwned = deferred_owned, drainCandidates = drain_candidates}) end if operation == 'inventory' then local count = tonumber(redis.call('HGET', inventory_key, upstream_id) or '0') diff --git a/internal/controller/health/unhealthy_reaper.go b/internal/controller/health/unhealthy_reaper.go index d09ce5a..684d081 100644 --- a/internal/controller/health/unhealthy_reaper.go +++ b/internal/controller/health/unhealthy_reaper.go @@ -53,9 +53,23 @@ func (reaper *ConfiguredUnhealthyReaper) Tick(ctx context.Context) (activitypool if len(policies) == 0 { return activitypool.UnhealthySweepResult{}, nil } - return reaper.store.SweepUnhealthy(ctx, activitypool.UnhealthySweepCommand{ - Now: now.UTC(), Limit: reaper.options.BatchSize, RemoveAfterByUpstream: policies, + utcNow := now.UTC() + result, err := reaper.store.SweepUnhealthy(ctx, activitypool.UnhealthySweepCommand{ + Now: utcNow, Limit: reaper.options.BatchSize, RemoveAfterByUpstream: policies, }) + if err != nil { + return activitypool.UnhealthySweepResult{}, err + } + drainer, supported := reaper.store.(activitypool.UnhealthyDrainStarter) + if !supported || nilInterface(drainer) { + return result, nil + } + for _, candidate := range result.DrainCandidates { + if _, err := drainer.BeginUnhealthyDrain(ctx, utcNow, candidate); err != nil { + return result, err + } + } + return result, nil } func (reaper *ConfiguredUnhealthyReaper) Run(ctx context.Context) error { diff --git a/internal/controller/health/unhealthy_reaper_test.go b/internal/controller/health/unhealthy_reaper_test.go index 744bf17..2d13db9 100644 --- a/internal/controller/health/unhealthy_reaper_test.go +++ b/internal/controller/health/unhealthy_reaper_test.go @@ -10,9 +10,23 @@ import ( ) type unhealthyRemoverStub struct { - command activitypool.UnhealthySweepCommand - calls int - result activitypool.UnhealthySweepResult + command activitypool.UnhealthySweepCommand + calls int + result activitypool.UnhealthySweepResult + drainCalls int + drainNow time.Time + candidates []activitypool.UnhealthyDrainCandidate +} + +func (store *unhealthyRemoverStub) BeginUnhealthyDrain( + _ context.Context, + now time.Time, + candidate activitypool.UnhealthyDrainCandidate, +) (bool, error) { + store.drainCalls++ + store.drainNow = now + store.candidates = append(store.candidates, candidate) + return true, nil } func (store *unhealthyRemoverStub) SweepUnhealthy( @@ -39,7 +53,7 @@ func TestConfiguredUnhealthyReaperUsesEffectiveEnabledPolicies(t *testing.T) { t.Fatalf("NewConfiguredUnhealthyReaper() = %v", err) } result, err := reaper.Tick(context.Background()) - if err != nil || result != (activitypool.UnhealthySweepResult{Removed: 2}) { + if err != nil || result.Removed != 2 || result.DeferredOwned != 0 || len(result.DrainCandidates) != 0 { t.Fatalf("Tick() = %+v, %v", result, err) } if store.calls != 1 || !store.command.Now.Equal(now) || store.command.Limit != 32 || @@ -58,7 +72,32 @@ func TestConfiguredUnhealthyReaperSkipsDisabledRemoval(t *testing.T) { if err != nil { t.Fatalf("NewConfiguredUnhealthyReaper() = %v", err) } - if result, err := reaper.Tick(context.Background()); err != nil || result != (activitypool.UnhealthySweepResult{}) || store.calls != 0 { + if result, err := reaper.Tick(context.Background()); err != nil || result.Removed != 0 || result.DeferredOwned != 0 || len(result.DrainCandidates) != 0 || store.calls != 0 { t.Fatalf("Tick(disabled) = (%+v, %v), calls=%d", result, err, store.calls) } } + +func TestConfiguredUnhealthyReaperStartsBoundedDrainCandidates(t *testing.T) { + now := time.Date(2026, 8, 2, 14, 0, 0, 0, time.UTC) + candidate := activitypool.UnhealthyDrainCandidate{ + ProxyID: "proxy-a", WorkerID: "worker-a", AssignmentEpoch: 7, UnhealthySince: now.Add(-time.Minute), + } + store := &unhealthyRemoverStub{result: activitypool.UnhealthySweepResult{ + DeferredOwned: 1, DrainCandidates: []activitypool.UnhealthyDrainCandidate{candidate}, + }} + reaper, err := NewConfiguredUnhealthyReaper(&configurationSourceStub{configuration: &config.Config{ + Upstreams: map[string]config.Upstream{ + "provider-a": {Enabled: true, Check: config.Check{UnhealthyRemoveAfter: config.Duration(time.Minute)}}, + }, + }}, store, UnhealthyReaperOptions{PollInterval: time.Second, BatchSize: 4, Now: func() time.Time { return now }}) + if err != nil { + t.Fatalf("NewConfiguredUnhealthyReaper() = %v", err) + } + result, err := reaper.Tick(context.Background()) + if err != nil || result.DeferredOwned != 1 || len(result.DrainCandidates) != 1 { + t.Fatalf("Tick() = %+v, %v", result, err) + } + if store.drainCalls != 1 || !store.drainNow.Equal(now) || len(store.candidates) != 1 || store.candidates[0] != candidate { + t.Fatalf("BeginUnhealthyDrain() calls=%d now=%v candidates=%+v", store.drainCalls, store.drainNow, store.candidates) + } +} diff --git a/internal/domain/activitypool/global_health_test.go b/internal/domain/activitypool/global_health_test.go index cc2f854..544ec87 100644 --- a/internal/domain/activitypool/global_health_test.go +++ b/internal/domain/activitypool/global_health_test.go @@ -40,7 +40,7 @@ func TestMemoryPoolAppliesGlobalObservationAtomically(t *testing.T) { } } -func TestMemoryPoolSweepsOnlyUnownedProxiesPastConfiguredUnhealthyGrace(t *testing.T) { +func TestMemoryPoolStartsConditionalDrainForOwnedProxyPastUnhealthyGrace(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{ @@ -71,20 +71,29 @@ func TestMemoryPoolSweepsOnlyUnownedProxiesPastConfiguredUnhealthyGrace(t *testi 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}) { + result, err := pool.SweepUnhealthy(context.Background(), command) + if err != nil || result.Removed != 0 || result.DeferredOwned != 1 || len(result.DrainCandidates) != 1 { t.Fatalf("SweepUnhealthy(owned) = %+v, %v", result, err) } + candidate := result.DrainCandidates[0] + if candidate.ProxyID != "proxy-a" || candidate.WorkerID != "worker-a" || candidate.AssignmentEpoch != assignment.Epoch || + !candidate.UnhealthySince.Equal(now.Add(5*time.Second)) { + t.Fatalf("SweepUnhealthy() candidate = %+v", candidate) + } 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 started, err := pool.BeginUnhealthyDrain(context.Background(), command.Now, candidate); err != nil || !started { + t.Fatalf("BeginUnhealthyDrain() = %t, %v", started, err) + } + if started, err := pool.BeginUnhealthyDrain(context.Background(), command.Now, candidate); err != nil || started { + t.Fatalf("BeginUnhealthyDrain(idempotent) = %t, %v", started, 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}) { + if result, err := pool.SweepUnhealthy(context.Background(), command); err != nil || result.Removed != 1 || result.DeferredOwned != 0 || len(result.DrainCandidates) != 0 { t.Fatalf("SweepUnhealthy(unowned) = %+v, %v", result, err) } if _, exists := pool.entryByIDLocked("proxy-a"); exists { @@ -92,6 +101,58 @@ func TestMemoryPoolSweepsOnlyUnownedProxiesPastConfiguredUnhealthyGrace(t *testi } } +func TestMemoryPoolRejectsUnhealthyDrainCandidateAfterRecovery(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: 1, RemoveAfterByUpstream: map[string]time.Duration{"provider-a": time.Second}, + } + result, err := pool.SweepUnhealthy(context.Background(), command) + if err != nil || len(result.DrainCandidates) != 1 { + t.Fatalf("SweepUnhealthy() = %+v, %v", result, err) + } + if _, err := pool.ApplyHealth(context.Background(), HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: now.Add(7 * time.Second), NextState: proxyDomain.StateChecking, + }); err != nil { + t.Fatalf("ApplyHealth(rechecking) = %v", err) + } + if _, err := pool.ApplyGlobalObservation(context.Background(), GlobalHealthCommand{ + Observation: healthObservation("task-recovered", true, now.Add(8*time.Second)), MaxConsecutiveFailures: 1, + }); err != nil { + t.Fatalf("ApplyGlobalObservation(recovered) = %v", err) + } + if started, err := pool.BeginUnhealthyDrain(context.Background(), now.Add(8*time.Second), result.DrainCandidates[0]); err != nil || started { + t.Fatalf("BeginUnhealthyDrain(recovered) = %t, %v", started, err) + } + if current, ok, err := pool.Get(context.Background(), "proxy-a"); err != nil || !ok || current.Epoch != assignment.Epoch || current.Draining { + t.Fatalf("Get(recovered) = %+v, %t, %v", current, ok, err) + } +} + 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 e6c53aa..39263cc 100644 --- a/internal/domain/activitypool/pool.go +++ b/internal/domain/activitypool/pool.go @@ -130,18 +130,38 @@ type UnhealthySweepCommand struct { 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. +// UnhealthyDrainCandidate is a bounded hint returned by an unhealthy sweep. +// Consumers must submit it to UnhealthyDrainStarter; it is not an authority +// to drain on its own because health and ownership can change after a sweep. +type UnhealthyDrainCandidate struct { + ProxyID string + WorkerID string + AssignmentEpoch uint64 + UnhealthySince time.Time +} + +// UnhealthySweepResult separates completed removals from owned entries. A +// bounded candidate list allows the Controller to begin a fenced Drain for +// unhealthy assignments without placing this maintenance work on the Gateway +// request path. type UnhealthySweepResult struct { - Removed int - DeferredOwned int + Removed int + DeferredOwned int + DrainCandidates []UnhealthyDrainCandidate } type UnhealthyRemover interface { SweepUnhealthy(context.Context, UnhealthySweepCommand) (UnhealthySweepResult, error) } +// UnhealthyDrainStarter conditionally begins a Drain for a candidate selected +// by SweepUnhealthy. It returns true only when this call changed an assignment +// from active ownership to draining; stale, recovered, reassigned, and already +// draining candidates return false without error. +type UnhealthyDrainStarter interface { + BeginUnhealthyDrain(context.Context, time.Time, UnhealthyDrainCandidate) (bool, error) +} + const maximumUnhealthySweepScan = 1024 type Entry struct { @@ -186,6 +206,7 @@ var ( _ StateInventoryReader = (*MemoryPool)(nil) _ Maintainer = (*MemoryPool)(nil) _ UnhealthyRemover = (*MemoryPool)(nil) + _ UnhealthyDrainStarter = (*MemoryPool)(nil) _ extractionDomain.Store = (*MemoryPool)(nil) _ ownershipDomain.Repository = (*MemoryPool)(nil) _ ownershipDomain.DrainTicketStore = (*MemoryPool)(nil) @@ -672,6 +693,13 @@ func (p *MemoryPool) SweepUnhealthy( // Retry later without losing the authoritative first-unhealthy timestamp. p.unhealthy[candidate.proxyID] = command.Now.UTC().Add(time.Second) result.DeferredOwned++ + if assignment, owned := p.ownership[candidate.proxyID]; owned && !assignment.Draining && + assignment.WorkerID == entry.OwnerWorkerID && len(result.DrainCandidates) < command.Limit { + result.DrainCandidates = append(result.DrainCandidates, UnhealthyDrainCandidate{ + ProxyID: candidate.proxyID, WorkerID: assignment.WorkerID, AssignmentEpoch: assignment.Epoch, + UnhealthySince: entry.GlobalHealth.UnhealthySince.UTC(), + }) + } continue } p.removeEntryByIDLocked(candidate.proxyID) @@ -882,19 +910,52 @@ func (p *MemoryPool) BeginDrain(ctx context.Context, proxyID, workerID string, e if !ok || assignment.WorkerID != workerID || assignment.Epoch != epoch { return ownershipDomain.Assignment{}, ownershipDomain.ErrStaleAssignment } - if !assignment.Draining { - assignment.Draining = true - assignment.Version++ - p.ownership[proxyID] = assignment - p.nextEpoch++ - p.drains[proxyID] = ownershipDomain.DrainTicket{ - ProxyID: proxyID, WorkerID: workerID, AssignmentEpoch: assignment.Epoch, - RequiredSnapshotEpoch: p.nextEpoch, - } - } + assignment, _ = p.beginDrainLocked(assignment) return assignment, nil } +func (p *MemoryPool) BeginUnhealthyDrain(ctx context.Context, now time.Time, candidate UnhealthyDrainCandidate) (bool, error) { + if err := ownershipContextError(ctx); err != nil { + return false, err + } + if p == nil || now.IsZero() || candidate.ProxyID == "" || candidate.WorkerID == "" || candidate.AssignmentEpoch == 0 || candidate.UnhealthySince.IsZero() { + return false, ErrInvalidMaintenance + } + p.mu.Lock() + defer p.mu.Unlock() + if err := ctx.Err(); err != nil { + return false, err + } + p.purgeExpiredLocked(now) + assignment, owned := p.ownership[candidate.ProxyID] + entry, exists := p.entryByIDLocked(candidate.ProxyID) + if !owned || !exists || assignment.WorkerID != candidate.WorkerID || assignment.Epoch != candidate.AssignmentEpoch || + entry.OwnerWorkerID != candidate.WorkerID || entry.State != proxyDomain.StateUnhealthy || + !entry.GlobalHealth.UnhealthySince.Equal(candidate.UnhealthySince) { + return false, nil + } + if assignment.Draining { + return false, nil + } + _, started := p.beginDrainLocked(assignment) + return started, nil +} + +func (p *MemoryPool) beginDrainLocked(assignment ownershipDomain.Assignment) (ownershipDomain.Assignment, bool) { + if assignment.Draining { + return assignment, false + } + assignment.Draining = true + assignment.Version++ + p.ownership[assignment.ProxyID] = assignment + p.nextEpoch++ + p.drains[assignment.ProxyID] = ownershipDomain.DrainTicket{ + ProxyID: assignment.ProxyID, WorkerID: assignment.WorkerID, AssignmentEpoch: assignment.Epoch, + RequiredSnapshotEpoch: p.nextEpoch, + } + return assignment, true +} + func (p *MemoryPool) PendingDrains(ctx context.Context, workerID string, limit int) ([]ownershipDomain.DrainTicket, error) { if err := ownershipContextError(ctx); err != nil { return nil, err diff --git a/progress.md b/progress.md index 6dc21d3..3ec7b24 100644 --- a/progress.md +++ b/progress.md @@ -10,8 +10,9 @@ `scripts/verify.ps1` 全部通过，后者按环境跳过 `CGO_ENABLED` 未启用的 race 测试。 - 已完成持续 `UNHEALTHY` 后的安全淘汰：首次异常时间写入 Redis TTL 活动池， `check.unhealthyRemoveAfter` 按 Upstream 生效；Controller 通过公用、有界的 - Reaper 仅回收未分配代理。仍持有 Worker ownership 的候选会延后，待既有 - Drain 收尾清空 ownership 后再删除，避免影响仍在 Gateway Snapshot 中使用的代理。 + Reaper 回收未分配代理。仍持有 Worker ownership 的候选以 `(proxy, worker, assignment + epoch, unhealthySince)` 有界返回，并由条件式 Drain 原子复核后创建 Ticket；排除 Snapshot + ACK 与 Runtime 零计数收尾后再删除，避免影响仍在 Gateway Snapshot 中使用的代理。 - Redis 集成标签测试在未设置 `PROXY_POOL_TEST_REDIS_URL` 时现能正确跳过所有契约 子测试；`go test ./...`、`go test -tags=integration ./internal/adapters/redisactivity` 与 `scripts/verify.ps1` 已通过，race 仍由 Linux CI 执行。 diff --git a/task_plan.md b/task_plan.md index b52434f..6873e33 100644 --- a/task_plan.md +++ b/task_plan.md @@ -39,11 +39,11 @@ 完成，WorkerControlPlane、Provider、业务指标与完整容器进程链仍待实现 14. [已完成] 收敛健康故障生命周期：将 `UNHEALTHY` 的持续时长纳入权威活动池状态， 通过 Controller 后台的有界回收清理无 Worker ownership 的持续异常 Proxy；拥有 - Worker 的候选延后到既有 Drain/ACK 清除所有权后再处理。 + Worker 的候选经 health/ownership 条件栅栏自动发起 Drain，在 ACK 与 Runtime 收尾后再处理。 15. [进行中] 收敛 Worker 发布生命周期：已完成权威 Proxy/Routing 的持续完整快照 刷新，以及 Drain 发起时的持久化 Ticket、Worker 待绑定索引、ownership epoch 推进和 完整排除 Snapshot 屏障绑定；已在 ACK 与 Runtime 零计数之间完成 Redis 原子自动收尾。 - 后续将 Reaper、配置停用与健康淘汰接入 `BeginDrain` 编排，形成可观测闭环。 + 后续将配置停用接入同一 `BeginDrain` 编排，并补齐可观测闭环。 ## 串并行关系