diff --git a/README.md b/README.md index 736a42e..f7305d4 100644 --- a/README.md +++ b/README.md @@ -263,7 +263,8 @@ go run ./cmd/proxy-loadgen ` 回收会自动为已分配项发起带健康/归属栅栏的 Drain，待 Snapshot ACK 与运行态归零后再删除； TARGET 事实按路由目标 Profile 独立归并。 - **P1 - Gateway 与 Routing**：Gateway 进程、快照凭据分发、五种 Routing 策略与 - `onUnavailable` 已接入；动态容量调整和配置停用触发的 Drain 编排待完成。 + `onUnavailable` 已接入；上游停用会从后续完整 Snapshot 排除，并对现有 Worker + ownership 发起带策略 revision 栅栏的 Drain。动态容量调整仍待完成。 - **P1 - 可观测与部署**：低基数业务指标、完整 Compose/Kubernetes 进程拓扑， 以及故障转移和恢复演练。 - **P2 - 容量证明**：`proxy-loadgen`、HTTP/CONNECT/Extract 分场景压测，以及 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index f00b4e5..cb3820b 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -302,6 +302,13 @@ Ticket/排除快照/Runtime 零计数闭环。 Controller 即可生效。Redis 任务存储现已扩展 BASIC/EGRESS/TARGET 的独立有界索引；路由目标 Profile 在启用 Routing 与 Upstream 的组合上才进入调度。 +补充进度（2026-08-02）：配置停用已进入 ownership Drain 编排。Controller 以静态配置 +与 Admin 管理态的交集生成完整、版本化的 Upstream 策略视图；Redis 只从对应 `owned` 索引 +有界读取仍可用的已归属 Proxy。候选携带 Upstream revision，`ownership.lua` 在复用既有 +Ticket/排除 Snapshot/Runtime 零计数闭环前，原子复核策略仍为停用、Proxy source、Worker、 +assignment epoch 和未过期租约。重启用后的策略 revision 会使旧候选返回无操作。Provider +Supervisor 也改为同时服从静态配置与管理态，消除两条启停消费链的不一致。 + 补充进度（2026-08-02）：已新增 `proxy-loadgen` HTTP 与 CONNECT 长连接场景。固定请求数 和固定时长两种模式均通过固定 worker 数与有界派发通道执行，可选 QPS 限速；报告使用固定大小 延迟直方图，输出状态分类、CONNECT 建立数、Extract 校验数、吞吐和 Go 内存/GC 快照。CONNECT diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index f989dde..7c97f67 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -10,7 +10,7 @@ | ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime/Outcome 与 Checker Observation gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Checker 任务流已具备有界领取、租约栅栏和任务期凭据契约。Gateway 已将快照编译为同版本动态 View，并由独立进程维护控制面会话。`proxy-checker` 与 Redis BASIC/EGRESS/TARGET 共享任务运行态、`proxy-loadgen` 有界 HTTP/CONNECT/Extract 校验场景均已实现 | | ARCH-002 | 热路径只做认证、本地路由和网络转发 | 1-70, 380-430 | Gateway bootstrap 集成测试验证启动期控制面会话与快照就绪，HTTP 请求只走本地 Snapshot/Dispatch；Outcome 仅写入有界非阻塞本地队列，代表性性能剖析待完成 | | ARCH-003 | Gateway、Distribution、Admin、Metrics 独立入口 | 8904-8958 | Controller 命令已装配 Distribution/Admin/Metrics 三个独立监听及联动停机；Gateway 命令已装配代理与 Metrics 监听，运行时 mTLS 部署 Overlay 待完成 | -| ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；Admin disable/reload 驱动取消替换，多副本按权威 HMAC 指纹和 revision 栅栏收敛并拒绝旧配置换主；Routing 切换到 Drain 的编排待完成 | +| ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；静态配置与 Admin 管理态共同决定 Provider 启停，停用策略以 Upstream revision 栅栏驱动既有 ownership Drain；多副本按权威 HMAC 指纹和 revision 栅栏收敛并拒绝旧配置换主；Routing 切换到 Drain 的编排待完成 | | ARCH-005 | 100k QPS 峰值使用多 Worker 集群 | 当前会话 | 未验证设计目标；待代表性集群负载报告 | ## Routing 与 Upstream diff --git a/findings.md b/findings.md index b879043..a2425b5 100644 --- a/findings.md +++ b/findings.md @@ -124,7 +124,9 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两 Gateway 使用既有 `SessionSupervisor` 退避重连。Gateway 本地 Store 对从完整 Snapshot 消失且仍有 Active/Reserved 的 Proxy 已按 draining 继续上报。Redis Runtime 替换已在当前 session ACK、Ticket 屏障和完整零计数同时成立时自动释放 owner；Reaper 已对持续 - `UNHEALTHY` 的已归属代理调用条件式 Drain。配置停用尚未接入同一触发编排。 + `UNHEALTHY` 的已归属代理调用条件式 Drain。配置停用也会发布有效的上游策略视图， + 从上游 owned 索引有界选择候选；候选携带上游 revision，Redis 在创建 Drain Ticket 前 + 再次核验策略仍停用、Proxy source、Worker 和 assignment epoch，避免旧候选跨启停使用。 - 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 8290146..0188ba8 100644 --- a/internal/adapters/redisactivity/global_health_integration_test.go +++ b/internal/adapters/redisactivity/global_health_integration_test.go @@ -224,6 +224,53 @@ func TestRedisRejectsUnhealthyDrainCandidateAfterRecovery(t *testing.T) { } } +func TestRedisDisabledUpstreamDrainFencesPolicyRevision(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), time.Minute, + testProxy("proxy-a", "192.0.2.10")) + assignment, err := fixture.Adapter.Assign(context.Background(), now.Add(2*time.Second), "proxy-a", "worker-a", time.Minute) + if err != nil { + t.Fatalf("Assign() error = %v", err) + } + disabled := activitypool.UpstreamDrainPolicy{UpstreamID: "provider-a", Revision: 31} + if err := fixture.Adapter.ReplaceUpstreamDrainPolicies(context.Background(), []activitypool.UpstreamDrainPolicy{disabled}); err != nil { + t.Fatalf("ReplaceUpstreamDrainPolicies(disabled) error = %v", err) + } + candidates, err := fixture.Adapter.ListDisabledUpstreamDrainCandidates(context.Background(), now.Add(3*time.Second), disabled, 1) + if err != nil || len(candidates) != 1 || candidates[0] != (activitypool.DisabledUpstreamDrainCandidate{ + UpstreamID: "provider-a", PolicyRevision: 31, ProxyID: "proxy-a", WorkerID: "worker-a", AssignmentEpoch: assignment.Epoch, + }) { + t.Fatalf("ListDisabledUpstreamDrainCandidates() = %+v, %v", candidates, err) + } + if err := fixture.Adapter.ReplaceUpstreamDrainPolicies(context.Background(), []activitypool.UpstreamDrainPolicy{{ + UpstreamID: "provider-a", Revision: 32, Enabled: true, + }}); err != nil { + t.Fatalf("ReplaceUpstreamDrainPolicies(enabled) error = %v", err) + } + if started, err := fixture.Adapter.BeginDisabledUpstreamDrain(context.Background(), now.Add(3*time.Second), candidates[0]); err != nil || started { + t.Fatalf("BeginDisabledUpstreamDrain(stale policy) = %t, %v", started, err) + } + if current, ok, err := fixture.Adapter.Get(context.Background(), "proxy-a"); err != nil || !ok || current.Draining { + t.Fatalf("Get(after stale candidate) = %+v, %t, %v", current, ok, err) + } + + disabled.Revision = 33 + if err := fixture.Adapter.ReplaceUpstreamDrainPolicies(context.Background(), []activitypool.UpstreamDrainPolicy{disabled}); err != nil { + t.Fatalf("ReplaceUpstreamDrainPolicies(disabled again) error = %v", err) + } + candidates, err = fixture.Adapter.ListDisabledUpstreamDrainCandidates(context.Background(), now.Add(4*time.Second), disabled, 1) + if err != nil || len(candidates) != 1 { + t.Fatalf("ListDisabledUpstreamDrainCandidates(disabled again) = %+v, %v", candidates, err) + } + if started, err := fixture.Adapter.BeginDisabledUpstreamDrain(context.Background(), now.Add(4*time.Second), candidates[0]); err != nil || !started { + t.Fatalf("BeginDisabledUpstreamDrain() = %t, %v", started, err) + } + if current, ok, err := fixture.Adapter.Get(context.Background(), "proxy-a"); err != nil || !ok || !current.Draining { + t.Fatalf("Get(after drain) = %+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/keys.go b/internal/adapters/redisactivity/keys.go index fc09f72..7b88342 100644 --- a/internal/adapters/redisactivity/keys.go +++ b/internal/adapters/redisactivity/keys.go @@ -9,67 +9,69 @@ import ( const redisKeyPrefix = "pp:{activity}:" type keyspace struct { - prefix string - records string - unique string - idkeys string - expiry string - available string - owners string - ownerExpiry string - drainTickets string - epoch string - inventory string - stateInventory string - workerSessions string - workerSessionExpiry string - workerSnapshots string - workerSnapshotExpiry string - workerRuntime string - workerRuntimeExpiry string - workerOutcomes string - healthDue string - healthUnhealthy string - healthEgressDue string - healthTargetDue string - healthQueued string - healthLeases string - healthTasks string - healthTaskExpiry string - healthRefTask string + prefix string + records string + unique string + idkeys string + expiry string + available string + owners string + ownerExpiry string + drainTickets string + upstreamDrainPolicies string + epoch string + inventory string + stateInventory string + workerSessions string + workerSessionExpiry string + workerSnapshots string + workerSnapshotExpiry string + workerRuntime string + workerRuntimeExpiry string + workerOutcomes string + healthDue string + healthUnhealthy string + healthEgressDue string + healthTargetDue string + healthQueued string + healthLeases string + healthTasks string + healthTaskExpiry string + healthRefTask string } func newKeyspace(namespace string) keyspace { prefix := redisKeyPrefix + namespace return keyspace{ - prefix: prefix, - records: prefix + ":records", - unique: prefix + ":unique", - idkeys: prefix + ":idkeys", - expiry: prefix + ":expiry", - available: prefix + ":available", - owners: prefix + ":owners", - ownerExpiry: prefix + ":owner-expiry", - drainTickets: prefix + ":drain-tickets", - epoch: prefix + ":epoch", - inventory: prefix + ":inventory", - stateInventory: prefix + ":state-inventory", - workerSessions: prefix + ":worker-sessions", - workerSessionExpiry: prefix + ":worker-session-expiry", - workerSnapshots: prefix + ":worker-snapshots", - workerSnapshotExpiry: prefix + ":worker-snapshot-expiry", - workerRuntime: prefix + ":worker-runtime", - 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", - healthLeases: prefix + ":health-leases", - healthTasks: prefix + ":health-tasks", - healthTaskExpiry: prefix + ":health-task-expiry", - healthRefTask: prefix + ":health-ref-task", + prefix: prefix, + records: prefix + ":records", + unique: prefix + ":unique", + idkeys: prefix + ":idkeys", + expiry: prefix + ":expiry", + available: prefix + ":available", + owners: prefix + ":owners", + ownerExpiry: prefix + ":owner-expiry", + drainTickets: prefix + ":drain-tickets", + upstreamDrainPolicies: prefix + ":upstream-drain-policies", + epoch: prefix + ":epoch", + inventory: prefix + ":inventory", + stateInventory: prefix + ":state-inventory", + workerSessions: prefix + ":worker-sessions", + workerSessionExpiry: prefix + ":worker-session-expiry", + workerSnapshots: prefix + ":worker-snapshots", + workerSnapshotExpiry: prefix + ":worker-snapshot-expiry", + workerRuntime: prefix + ":worker-runtime", + 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", + healthLeases: prefix + ":health-leases", + healthTasks: prefix + ":health-tasks", + healthTaskExpiry: prefix + ":health-task-expiry", + healthRefTask: prefix + ":health-ref-task", } } diff --git a/internal/adapters/redisactivity/maintenance.go b/internal/adapters/redisactivity/maintenance.go index 9de1e0d..05ca128 100644 --- a/internal/adapters/redisactivity/maintenance.go +++ b/internal/adapters/redisactivity/maintenance.go @@ -3,22 +3,25 @@ package redisactivity import ( "context" "encoding/json" + "strconv" "time" "proxy-pool/internal/domain/activitypool" ) const ( - maintenanceInventory = "inventory" - maintenanceSweep = "sweep" - maintenanceUnhealthy = "unhealthy" + maintenanceInventory = "inventory" + maintenanceSweep = "sweep" + maintenanceUnhealthy = "unhealthy" + maintenanceDisabledUpstream = "disabled_upstream" maximumUnhealthySweepScan = 1024 ) var ( - _ activitypool.InventoryReader = (*Adapter)(nil) - _ activitypool.Maintainer = (*Adapter)(nil) + _ activitypool.InventoryReader = (*Adapter)(nil) + _ activitypool.Maintainer = (*Adapter)(nil) + _ activitypool.DisabledUpstreamDrainLister = (*Adapter)(nil) ) func (a *Adapter) Inventory(ctx context.Context, upstreamID string, now time.Time) (activitypool.Inventory, error) { @@ -124,6 +127,52 @@ func (a *Adapter) SweepUnhealthy( return result, nil } +func (a *Adapter) ListDisabledUpstreamDrainCandidates( + ctx context.Context, + now time.Time, + policy activitypool.UpstreamDrainPolicy, + limit int, +) ([]activitypool.DisabledUpstreamDrainCandidate, error) { + if ctx == nil { + return nil, activitypool.ErrInvalidMaintenance + } + if err := ctx.Err(); err != nil { + return nil, err + } + if a == nil || now.IsZero() || policy.UpstreamID == "" || policy.Revision == 0 || policy.Enabled || limit <= 0 { + return nil, activitypool.ErrInvalidMaintenance + } + if limit > maximumUnhealthySweepScan/4 { + limit = maximumUnhealthySweepScan / 4 + } + reply, err := a.runMaintenance(ctx, maintenanceDisabledUpstream, now, limit, policy.UpstreamID, "") + if err != nil { + return nil, err + } + if reply.Status == scriptStale { + return nil, nil + } + if reply.Status == scriptInvalid { + return nil, activitypool.ErrInvalidMaintenance + } + if reply.Status != scriptOK || len(reply.DisabledDrainCandidates) > limit { + return nil, invalidScriptReply("unexpected disabled upstream sweep reply") + } + wantRevision := strconv.FormatUint(policy.Revision, 10) + result := make([]activitypool.DisabledUpstreamDrainCandidate, 0, len(reply.DisabledDrainCandidates)) + for _, candidate := range reply.DisabledDrainCandidates { + if candidate.UpstreamID != policy.UpstreamID || candidate.PolicyRevision != wantRevision || + candidate.ProxyID == "" || candidate.WorkerID == "" || candidate.AssignmentEpoch == 0 { + return nil, invalidScriptReply("disabled upstream sweep reply contained an invalid drain candidate") + } + result = append(result, activitypool.DisabledUpstreamDrainCandidate{ + UpstreamID: candidate.UpstreamID, PolicyRevision: policy.Revision, ProxyID: candidate.ProxyID, + WorkerID: candidate.WorkerID, AssignmentEpoch: candidate.AssignmentEpoch, + }) + } + return result, nil +} + func (a *Adapter) runMaintenance( ctx context.Context, operation string, @@ -141,6 +190,7 @@ func (a *Adapter) runMaintenance( 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, + a.keys.owned(upstreamID), a.keys.upstreamDrainPolicies, }, 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 66c8a93..ee44c81 100644 --- a/internal/adapters/redisactivity/ownership.go +++ b/internal/adapters/redisactivity/ownership.go @@ -6,6 +6,7 @@ import ( "encoding/hex" "errors" "fmt" + "strconv" "time" "proxy-pool/internal/domain/activitypool" @@ -17,6 +18,7 @@ const ( ownershipRenew = "renew" ownershipBeginDrain = "begin_drain" ownershipBeginUnhealthyDrain = "begin_unhealthy_drain" + ownershipBeginDisabledDrain = "begin_disabled_upstream_drain" ownershipAcknowledgeDrain = "acknowledge_drain" ownershipGet = "get" ownershipExpire = "expire" @@ -25,6 +27,7 @@ const ( var _ ownershipDomain.Repository = (*Adapter)(nil) var _ ownershipDomain.DrainTicketStore = (*Adapter)(nil) var _ activitypool.UnhealthyDrainStarter = (*Adapter)(nil) +var _ activitypool.DisabledUpstreamDrainStarter = (*Adapter)(nil) func (a *Adapter) Assign( ctx context.Context, @@ -39,7 +42,7 @@ func (a *Adapter) Assign( if now.IsZero() || proxyID == "" || workerID == "" || ttl <= 0 { return ownershipDomain.Assignment{}, ownershipDomain.ErrInvalidOwnership } - reply, err := a.runOwnership(ctx, ownershipAssign, true, now.UnixMilli(), proxyID, workerID, 0, durationMillis(ttl), 0, 0) + reply, err := a.runOwnership(ctx, ownershipAssign, true, now.UnixMilli(), proxyID, workerID, 0, durationMillis(ttl), 0, 0, "", "") if err != nil { return ownershipDomain.Assignment{}, err } @@ -71,7 +74,7 @@ func (a *Adapter) Renew( if now.IsZero() || proxyID == "" || workerID == "" || epoch == 0 || ttl <= 0 { return ownershipDomain.Assignment{}, ownershipDomain.ErrInvalidOwnership } - reply, err := a.runOwnership(ctx, ownershipRenew, true, now.UnixMilli(), proxyID, workerID, epoch, durationMillis(ttl), 0, 0) + reply, err := a.runOwnership(ctx, ownershipRenew, true, now.UnixMilli(), proxyID, workerID, epoch, durationMillis(ttl), 0, 0, "", "") if err != nil { return ownershipDomain.Assignment{}, err } @@ -99,7 +102,7 @@ func (a *Adapter) BeginDrain( if proxyID == "" || workerID == "" || epoch == 0 { return ownershipDomain.Assignment{}, ownershipDomain.ErrInvalidOwnership } - reply, err := a.runOwnership(ctx, ownershipBeginDrain, true, 0, proxyID, workerID, epoch, 0, 0, 0) + reply, err := a.runOwnership(ctx, ownershipBeginDrain, true, 0, proxyID, workerID, epoch, 0, 0, 0, "", "") if err != nil { return ownershipDomain.Assignment{}, err } @@ -128,7 +131,7 @@ func (a *Adapter) BeginUnhealthyDrain( } reply, err := a.runOwnership( ctx, ownershipBeginUnhealthyDrain, true, now.UTC().UnixMilli(), candidate.ProxyID, candidate.WorkerID, - candidate.AssignmentEpoch, candidate.UnhealthySince.UTC().UnixMilli(), 0, 0, + candidate.AssignmentEpoch, candidate.UnhealthySince.UTC().UnixMilli(), 0, 0, "", "", ) if err != nil { return false, err @@ -148,6 +151,40 @@ func (a *Adapter) BeginUnhealthyDrain( } } +func (a *Adapter) BeginDisabledUpstreamDrain( + ctx context.Context, + now time.Time, + candidate activitypool.DisabledUpstreamDrainCandidate, +) (bool, error) { + if err := validateOwnershipCall(ctx, a); err != nil { + return false, err + } + if now.IsZero() || candidate.UpstreamID == "" || candidate.PolicyRevision == 0 || candidate.ProxyID == "" || + candidate.WorkerID == "" || candidate.AssignmentEpoch == 0 { + return false, activitypool.ErrInvalidMaintenance + } + reply, err := a.runOwnership( + ctx, ownershipBeginDisabledDrain, true, now.UTC().UnixMilli(), candidate.ProxyID, candidate.WorkerID, + candidate.AssignmentEpoch, 0, 0, 0, candidate.UpstreamID, strconv.FormatUint(candidate.PolicyRevision, 10), + ) + 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 disabled upstream 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 @@ -242,7 +279,7 @@ func (a *Adapter) AcknowledgeDrain( if proxyID == "" || workerID == "" || epoch == 0 || active < 0 || reserved < 0 { return ownershipDomain.ErrInvalidOwnership } - reply, err := a.runOwnership(ctx, ownershipAcknowledgeDrain, true, 0, proxyID, workerID, epoch, 0, active, reserved) + reply, err := a.runOwnership(ctx, ownershipAcknowledgeDrain, true, 0, proxyID, workerID, epoch, 0, active, reserved, "", "") if err != nil { return err } @@ -269,7 +306,7 @@ func (a *Adapter) Get(ctx context.Context, proxyID string) (ownershipDomain.Assi if proxyID == "" { return ownershipDomain.Assignment{}, false, ownershipDomain.ErrInvalidOwnership } - reply, err := a.runOwnership(ctx, ownershipGet, false, 0, proxyID, "", 0, 0, 0, 0) + reply, err := a.runOwnership(ctx, ownershipGet, false, 0, proxyID, "", 0, 0, 0, 0, "", "") if err != nil { return ownershipDomain.Assignment{}, false, err } @@ -293,7 +330,7 @@ func (a *Adapter) Expire(ctx context.Context, now time.Time, limit int) ([]owner if now.IsZero() || limit <= 0 { return nil, ownershipDomain.ErrInvalidOwnership } - reply, err := a.runOwnership(ctx, ownershipExpire, true, now.UnixMilli(), "", "", 0, int64(limit), 0, 0) + reply, err := a.runOwnership(ctx, ownershipExpire, true, now.UnixMilli(), "", "", 0, int64(limit), 0, 0, "", "") if err != nil { return nil, err } @@ -331,6 +368,8 @@ func (a *Adapter) runOwnership( value int64, active int64, reserved int64, + upstreamID string, + policyRevision string, ) (ownershipScriptReply, error) { operationKey := a.keys.epoch if mutating { @@ -345,9 +384,9 @@ func (a *Adapter) runOwnership( 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.healthUnhealthy, - a.keys.drainTickets, + a.keys.drainTickets, a.keys.upstreamDrainPolicies, }, operation, operationTTLMillis(a.options.OperationTTL), a.options.CleanupLimit, - nowMS, proxyID, workerID, epoch, value, active, reserved, a.keys.workerOwned(workerID), a.keys.workerDraining(workerID)) + nowMS, proxyID, workerID, epoch, value, active, reserved, a.keys.workerOwned(workerID), a.keys.workerDraining(workerID), upstreamID, policyRevision) if err != nil { return ownershipScriptReply{}, err } diff --git a/internal/adapters/redisactivity/scripts.go b/internal/adapters/redisactivity/scripts.go index 8b5b044..536925e 100644 --- a/internal/adapters/redisactivity/scripts.go +++ b/internal/adapters/redisactivity/scripts.go @@ -88,16 +88,21 @@ type ownershipScriptReply struct { Started bool `json:"started"` } +type upstreamDrainPolicyScriptReply struct { + Status scriptStatus `json:"status"` +} + type drainTicketsScriptReply struct { Status scriptStatus `json:"status"` Tickets []string `json:"tickets"` } type maintenanceScriptReply struct { - Status scriptStatus `json:"status"` - Count int `json:"count"` - DeferredOwned int `json:"deferredOwned"` - DrainCandidates []maintenanceDrainCandidateScriptRow `json:"drainCandidates"` + Status scriptStatus `json:"status"` + Count int `json:"count"` + DeferredOwned int `json:"deferredOwned"` + DrainCandidates []maintenanceDrainCandidateScriptRow `json:"drainCandidates"` + DisabledDrainCandidates []disabledDrainCandidateScriptRow `json:"disabledDrainCandidates"` } type maintenanceDrainCandidateScriptRow struct { @@ -107,6 +112,14 @@ type maintenanceDrainCandidateScriptRow struct { UnhealthySinceMS int64 `json:"unhealthySinceMs"` } +type disabledDrainCandidateScriptRow struct { + UpstreamID string `json:"upstreamId"` + PolicyRevision string `json:"policyRevision"` + ProxyID string `json:"proxyId"` + WorkerID string `json:"workerId"` + AssignmentEpoch uint64 `json:"assignmentEpoch"` +} + type statusScriptReply struct { Status scriptStatus `json:"status"` Inventories []statusScriptInventory `json:"inventories"` @@ -188,21 +201,25 @@ var capacitySource string //go:embed scripts/worker_snapshot.lua var workerSnapshotSource string +//go:embed scripts/upstream_drain_policy.lua +var upstreamDrainPolicySource string + var ( - upsertScript = redis.NewScript(upsertSource) - healthScript = redis.NewScript(healthSource) - healthTasksScript = redis.NewScript(healthTasksSource) - targetHealthScript = redis.NewScript(targetHealthSource) - upstreamLookupScript = redis.NewScript(upstreamLookupSource) - extractScript = redis.NewScript(extractSource) - ownershipScript = redis.NewScript(ownershipSource) - drainTicketsScript = redis.NewScript(drainTicketsSource) - bindDrainTicketScript = redis.NewScript(bindDrainTicketSource) - sweepScript = redis.NewScript(sweepSource) - statusScript = redis.NewScript(statusSource) - runtimeScript = redis.NewScript(runtimeSource) - capacityScript = redis.NewScript(capacitySource) - workerSnapshotScript = redis.NewScript(workerSnapshotSource) + upsertScript = redis.NewScript(upsertSource) + healthScript = redis.NewScript(healthSource) + healthTasksScript = redis.NewScript(healthTasksSource) + targetHealthScript = redis.NewScript(targetHealthSource) + upstreamLookupScript = redis.NewScript(upstreamLookupSource) + extractScript = redis.NewScript(extractSource) + ownershipScript = redis.NewScript(ownershipSource) + drainTicketsScript = redis.NewScript(drainTicketsSource) + bindDrainTicketScript = redis.NewScript(bindDrainTicketSource) + sweepScript = redis.NewScript(sweepSource) + statusScript = redis.NewScript(statusSource) + runtimeScript = redis.NewScript(runtimeSource) + capacityScript = redis.NewScript(capacitySource) + workerSnapshotScript = redis.NewScript(workerSnapshotSource) + upstreamDrainPolicyScript = redis.NewScript(upstreamDrainPolicySource) ) func runScript(ctx context.Context, client redis.Scripter, script *redis.Script, keys []string, args ...any) (any, error) { diff --git a/internal/adapters/redisactivity/scripts/ownership.lua b/internal/adapters/redisactivity/scripts/ownership.lua index c584700..704b573 100644 --- a/internal/adapters/redisactivity/scripts/ownership.lua +++ b/internal/adapters/redisactivity/scripts/ownership.lua @@ -17,6 +17,7 @@ local health_task_expiry_key = KEYS[16] local health_ref_task_key = KEYS[17] local health_unhealthy_key = KEYS[18] local drain_tickets_key = KEYS[19] +local upstream_drain_policies_key = KEYS[20] local operation = ARGV[1] local operation_ttl_ms = tonumber(ARGV[2]) @@ -30,6 +31,8 @@ local active = tonumber(ARGV[9]) local reserved = tonumber(ARGV[10]) local worker_index_key = ARGV[11] local worker_draining_index_key = ARGV[12] +local upstream_id = ARGV[13] +local policy_revision = ARGV[14] local mutating = operation ~= 'get' local function finish(reply) @@ -212,6 +215,11 @@ local function decode_table(raw) return value end +local function valid_uint(value) + return type(value) == 'string' and string.match(value, '^[0-9]+$') and + value ~= '0' and (string.len(value) == 1 or string.sub(value, 1, 1) ~= '0') +end + local function valid_assignment(assignment) return assignment and assignment.version == 1 and type(assignment.proxyId) == 'string' and assignment.proxyId ~= '' and type(assignment.workerId) == 'string' and assignment.workerId ~= '' and @@ -396,6 +404,33 @@ if operation == 'begin_unhealthy_drain' then return finish({status = 'ok', record = encoded, started = started}) end +if operation == 'begin_disabled_upstream_drain' then + cleanup_hard_expired(now_ms) + if type(upstream_id) ~= 'string' or upstream_id == '' or not valid_uint(policy_revision) then + return finish({status = 'invalid'}) + end + local policy = decode_table(redis.call('HGET', upstream_drain_policies_key, upstream_id)) + if not policy or policy.version ~= 1 or policy.upstreamId ~= upstream_id or policy.enabled ~= false or + policy.revision ~= policy_revision then + return finish({status = 'stale'}) + end + 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.sourceUpstream ~= upstream_id or record.ownerWorkerId ~= worker_id or + record.state ~= 'AVAILABLE' 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 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 then diff --git a/internal/adapters/redisactivity/scripts/sweep.lua b/internal/adapters/redisactivity/scripts/sweep.lua index fd60aa5..9d70e5d 100644 --- a/internal/adapters/redisactivity/scripts/sweep.lua +++ b/internal/adapters/redisactivity/scripts/sweep.lua @@ -15,6 +15,8 @@ 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 owned_upstream_key = KEYS[18] +local upstream_drain_policies_key = KEYS[19] local operation = ARGV[1] local now_ms = tonumber(ARGV[2]) @@ -220,6 +222,40 @@ if operation == 'unhealthy' then end return finish({status = 'ok', count = removed, deferredOwned = deferred_owned, drainCandidates = drain_candidates}) end +if operation == 'disabled_upstream' then + local raw_policy = redis.call('HGET', upstream_drain_policies_key, upstream_id) + local policy = raw_policy and decode_table(raw_policy) or nil + if not policy or policy.version ~= 1 or policy.upstreamId ~= upstream_id or policy.enabled ~= false or + type(policy.revision) ~= 'string' or not string.match(policy.revision, '^[0-9]+$') or policy.revision == '0' then + return finish({status = 'stale', count = 0, disabledDrainCandidates = {}}) + end + local scan_limit = limit * 4 + if scan_limit > 1024 then + scan_limit = 1024 + end + local proxy_ids = redis.call('ZRANGEBYSCORE', owned_upstream_key, '(' .. now_ms, '+inf', 'LIMIT', 0, scan_limit) + local candidates = {} + for _, proxy_id in ipairs(proxy_ids) do + local record = decode_table(redis.call('HGET', records_key, proxy_id)) + local owner = decode_table(redis.call('HGET', owners_key, proxy_id)) + if record and owner and record.sourceUpstream == upstream_id and record.state == 'AVAILABLE' and + record.ownerWorkerId == owner.workerId and owner.draining == false and + type(owner.workerId) == 'string' and owner.workerId ~= '' and tonumber(owner.epoch) and + tonumber(owner.epoch) > 0 and tonumber(owner.expiresAtMs) and tonumber(owner.expiresAtMs) > now_ms then + table.insert(candidates, { + upstreamId = upstream_id, + policyRevision = policy.revision, + proxyId = proxy_id, + workerId = owner.workerId, + assignmentEpoch = tonumber(owner.epoch), + }) + if #candidates == limit then + break + end + end + end + return finish({status = 'ok', count = 0, disabledDrainCandidates = candidates}) +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/upstream_drain_policy.lua b/internal/adapters/redisactivity/scripts/upstream_drain_policy.lua new file mode 100644 index 0000000..cb165db --- /dev/null +++ b/internal/adapters/redisactivity/scripts/upstream_drain_policy.lua @@ -0,0 +1,23 @@ +local policies_key = KEYS[1] +local policies_json = ARGV[1] + +local decoded, policies = pcall(cjson.decode, policies_json or '') +if not decoded or type(policies) ~= 'table' then + return cjson.encode({status = 'invalid'}) +end + +local seen = {} +for _, policy in ipairs(policies) do + if type(policy) ~= 'table' or type(policy.upstreamId) ~= 'string' or policy.upstreamId == '' or + type(policy.revision) ~= 'string' or not string.match(policy.revision, '^[0-9]+$') or + policy.revision == '0' or type(policy.enabled) ~= 'boolean' or seen[policy.upstreamId] then + return cjson.encode({status = 'invalid'}) + end + seen[policy.upstreamId] = true +end + +redis.call('DEL', policies_key) +for _, policy in ipairs(policies) do + redis.call('HSET', policies_key, policy.upstreamId, cjson.encode(policy)) +end +return cjson.encode({status = 'ok'}) diff --git a/internal/adapters/redisactivity/upstream_drain_policy.go b/internal/adapters/redisactivity/upstream_drain_policy.go new file mode 100644 index 0000000..e42cba2 --- /dev/null +++ b/internal/adapters/redisactivity/upstream_drain_policy.go @@ -0,0 +1,63 @@ +package redisactivity + +import ( + "context" + "encoding/json" + "strconv" + + "proxy-pool/internal/domain/activitypool" +) + +var _ activitypool.UpstreamDrainPolicyWriter = (*Adapter)(nil) + +type upstreamDrainPolicyRecord struct { + Version int `json:"version"` + UpstreamID string `json:"upstreamId"` + Revision string `json:"revision"` + Enabled bool `json:"enabled"` +} + +func (a *Adapter) ReplaceUpstreamDrainPolicies(ctx context.Context, policies []activitypool.UpstreamDrainPolicy) error { + if ctx == nil { + return activitypool.ErrInvalidMaintenance + } + if err := ctx.Err(); err != nil { + return err + } + if a == nil { + return activitypool.ErrInvalidMaintenance + } + records := make([]upstreamDrainPolicyRecord, 0, len(policies)) + seen := make(map[string]struct{}, len(policies)) + for _, policy := range policies { + if policy.UpstreamID == "" || policy.Revision == 0 { + return activitypool.ErrInvalidMaintenance + } + if _, duplicate := seen[policy.UpstreamID]; duplicate { + return activitypool.ErrInvalidMaintenance + } + seen[policy.UpstreamID] = struct{}{} + records = append(records, upstreamDrainPolicyRecord{ + Version: 1, UpstreamID: policy.UpstreamID, Revision: strconv.FormatUint(policy.Revision, 10), Enabled: policy.Enabled, + }) + } + payload, err := json.Marshal(records) + if err != nil { + return err + } + result, err := runScript(ctx, a.client, upstreamDrainPolicyScript, []string{a.keys.upstreamDrainPolicies}, string(payload)) + if err != nil { + return err + } + var reply upstreamDrainPolicyScriptReply + if err := decodeScriptResult(result, &reply); err != nil { + return err + } + if reply.Status == scriptInvalid { + return activitypool.ErrInvalidMaintenance + } + if reply.Status != scriptOK { + return invalidScriptReply("unexpected upstream drain policy reply") + } + return nil +} diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index d486144..2ef906b 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -339,6 +339,30 @@ func runWithWorkerFactory( } runners = append(runners, reaper) } + if drainStore, ok := opened.activity.(controllerHealth.UpstreamDrainStore); ok && !nilInterface(drainStore) { + var policyStates []controllerHealth.UpstreamDrainStateSource + if loaded.Value.Admin.Enabled { + if nilInterface(opened.state) { + return errors.Join(ErrStartup, ErrInvalidOptions) + } + policyStates = append(policyStates, opened.state) + } + policySource, policyErr := controllerHealth.NewEffectiveUpstreamDrainPolicySource(configurationStore, policyStates...) + if policyErr != nil { + return fmt.Errorf("%w: build upstream Drain policy source: %w", ErrStartup, policyErr) + } + reaper, reaperErr := controllerHealth.NewConfiguredUpstreamDrainReaper( + policySource, + drainStore, + controllerHealth.UpstreamDrainReaperOptions{ + PollInterval: checkSchedulerPollInterval, BatchSize: checkSchedulerBatchSize, Now: options.Now, + }, + ) + if reaperErr != nil { + return fmt.Errorf("%w: build upstream Drain reaper: %w", ErrStartup, reaperErr) + } + runners = append(runners, reaper) + } } runners = append(runners, supervisor) group, err := lifecycle.NewGroup(runners...) diff --git a/internal/controller/bootstrap/provider_supervisor.go b/internal/controller/bootstrap/provider_supervisor.go index a893628..742fac1 100644 --- a/internal/controller/bootstrap/provider_supervisor.go +++ b/internal/controller/bootstrap/provider_supervisor.go @@ -276,7 +276,7 @@ func (supervisor *providerSupervisor) desired(ctx context.Context) (map[string]c for name, upstream := range configuration.Upstreams { isEnabled := upstream.Enabled if !nilInterface(supervisor.state) { - isEnabled = enabled[name] + isEnabled = isEnabled && enabled[name] } if isEnabled { upstream.Enabled = true diff --git a/internal/controller/bootstrap/provider_supervisor_test.go b/internal/controller/bootstrap/provider_supervisor_test.go index 0c30a2a..d9a0961 100644 --- a/internal/controller/bootstrap/provider_supervisor_test.go +++ b/internal/controller/bootstrap/provider_supervisor_test.go @@ -71,6 +71,38 @@ func TestProviderSupervisorAppliesDisableAndConfigurationReplacement(t *testing. } } +func TestProviderSupervisorKeepsStaticallyDisabledUpstreamStopped(t *testing.T) { + configuration, err := config.Load(strings.NewReader(bootstrapTestConfig)) + if err != nil { + t.Fatalf("config.Load(): %v", err) + } + disabled := configuration.Upstreams["provider-b"] + disabled.Enabled = false + configuration.Upstreams["provider-b"] = disabled + store, err := config.NewStore(configuration) + if err != nil { + t.Fatalf("config.NewStore(): %v", err) + } + supervisor, err := newProviderSupervisor( + store, + &mutableProviderState{enabled: map[string]bool{"provider-a": true, "provider-b": true}}, + func(string, config.Upstream) (lifecycle.Runner, error) { + return supervisorRunnerFunc(func(context.Context) error { return nil }), nil + }, + nil, nil, nil, bootstrapTestFingerprintKey, time.Hour, + ) + if err != nil { + t.Fatalf("newProviderSupervisor() error = %v", err) + } + desired, err := supervisor.desired(context.Background()) + if err != nil || len(desired) != 1 { + t.Fatalf("desired() = %+v, %v", desired, err) + } + if _, exists := desired["provider-a"]; !exists { + t.Fatalf("desired() = %+v, provider-a missing", desired) + } +} + func TestProviderSupervisorPropagatesUnexpectedRuntimeFailure(t *testing.T) { configuration, err := config.Load(strings.NewReader(bootstrapTestConfig)) if err != nil { diff --git a/internal/controller/health/upstream_drain_reaper.go b/internal/controller/health/upstream_drain_reaper.go new file mode 100644 index 0000000..00f24a0 --- /dev/null +++ b/internal/controller/health/upstream_drain_reaper.go @@ -0,0 +1,206 @@ +package health + +import ( + "context" + "errors" + "sort" + "time" + + "proxy-pool/internal/config" + "proxy-pool/internal/domain/activitypool" + "proxy-pool/internal/domain/adminstate" +) + +var ErrInvalidUpstreamDrainReaper = errors.New("invalid upstream drain reaper") + +// UpstreamDrainConfigurationSource returns one configuration and revision from +// the same atomic publication. config.Store implements this interface. +type UpstreamDrainConfigurationSource interface { + Snapshot() (*config.Config, uint64) +} + +type UpstreamDrainStateSource interface { + Snapshot(context.Context) (adminstate.Snapshot, error) +} + +// EffectiveUpstreamDrainPolicySource merges static configuration with optional +// Admin state. With Admin enabled it uses the same revision equality rule as +// Worker routing, so a torn config/state view produces no policy publication. +type EffectiveUpstreamDrainPolicySource struct { + configuration UpstreamDrainConfigurationSource + state UpstreamDrainStateSource +} + +func NewEffectiveUpstreamDrainPolicySource( + configuration UpstreamDrainConfigurationSource, + states ...UpstreamDrainStateSource, +) (*EffectiveUpstreamDrainPolicySource, error) { + if nilInterface(configuration) || len(states) > 1 || (len(states) == 1 && nilInterface(states[0])) { + return nil, ErrInvalidUpstreamDrainReaper + } + source := &EffectiveUpstreamDrainPolicySource{configuration: configuration} + if len(states) == 1 { + source.state = states[0] + } + return source, nil +} + +func (source *EffectiveUpstreamDrainPolicySource) ReadUpstreamDrainPolicies( + ctx context.Context, +) ([]activitypool.UpstreamDrainPolicy, error) { + if ctx == nil || source == nil || nilInterface(source.configuration) { + return nil, ErrInvalidUpstreamDrainReaper + } + if err := ctx.Err(); err != nil { + return nil, err + } + configuration, revision := source.configuration.Snapshot() + if configuration == nil { + return nil, ErrInvalidUpstreamDrainReaper + } + states := make(map[string]adminstate.UpstreamState, len(configuration.Upstreams)) + if source.state != nil { + snapshot, err := source.state.Snapshot(ctx) + if err != nil { + return nil, err + } + if snapshot.Config == nil || snapshot.Config.Revision != revision { + return nil, ErrInvalidUpstreamDrainReaper + } + for _, upstream := range snapshot.Upstreams { + if upstream.Name == "" || upstream.Revision == 0 { + return nil, ErrInvalidUpstreamDrainReaper + } + if _, duplicate := states[upstream.Name]; duplicate { + return nil, ErrInvalidUpstreamDrainReaper + } + states[upstream.Name] = upstream + } + } + if revision == 0 { + // A config-only Controller has no persisted management revision. Its + // configuration cannot be hot-reloaded, so one stable non-zero fence is enough. + revision = 1 + } + names := make([]string, 0, len(configuration.Upstreams)) + for name := range configuration.Upstreams { + names = append(names, name) + } + sort.Strings(names) + policies := make([]activitypool.UpstreamDrainPolicy, 0, len(names)) + for _, name := range names { + upstream := configuration.Upstreams[name] + policy := activitypool.UpstreamDrainPolicy{UpstreamID: name, Revision: revision, Enabled: upstream.Enabled} + if source.state != nil { + admin, exists := states[name] + if !exists { + return nil, ErrInvalidUpstreamDrainReaper + } + policy.Enabled = policy.Enabled && admin.Enabled + policy.Revision = admin.Revision + } + policies = append(policies, policy) + } + return policies, nil +} + +type UpstreamDrainPolicySource interface { + ReadUpstreamDrainPolicies(context.Context) ([]activitypool.UpstreamDrainPolicy, error) +} + +// UpstreamDrainStore is the narrow activity-pool capability required by the +// configuration-driven Drain reaper. +type UpstreamDrainStore interface { + activitypool.UpstreamDrainPolicyWriter + activitypool.DisabledUpstreamDrainLister + activitypool.DisabledUpstreamDrainStarter +} + +type UpstreamDrainReaperOptions struct { + PollInterval time.Duration + BatchSize int + Now func() time.Time +} + +type UpstreamDrainResult struct { + Candidates int + Started int +} + +// ConfiguredUpstreamDrainReaper continually publishes an effective, complete +// policy view before reading one bounded batch of disabled Upstream ownership. +// The Redis ownership operation performs the final policy-revision fence. +type ConfiguredUpstreamDrainReaper struct { + policies UpstreamDrainPolicySource + store UpstreamDrainStore + options UpstreamDrainReaperOptions +} + +func NewConfiguredUpstreamDrainReaper( + policies UpstreamDrainPolicySource, + store UpstreamDrainStore, + options UpstreamDrainReaperOptions, +) (*ConfiguredUpstreamDrainReaper, error) { + if nilInterface(policies) || nilInterface(store) || options.PollInterval <= 0 || options.BatchSize <= 0 || options.Now == nil { + return nil, ErrInvalidUpstreamDrainReaper + } + return &ConfiguredUpstreamDrainReaper{policies: policies, store: store, options: options}, nil +} + +func (reaper *ConfiguredUpstreamDrainReaper) Tick(ctx context.Context) (UpstreamDrainResult, error) { + if ctx == nil || reaper == nil || nilInterface(reaper.policies) || nilInterface(reaper.store) || + reaper.options.PollInterval <= 0 || reaper.options.BatchSize <= 0 || reaper.options.Now == nil { + return UpstreamDrainResult{}, ErrInvalidUpstreamDrainReaper + } + if err := ctx.Err(); err != nil { + return UpstreamDrainResult{}, err + } + now := reaper.options.Now() + if now.IsZero() { + return UpstreamDrainResult{}, ErrInvalidUpstreamDrainReaper + } + policies, err := reaper.policies.ReadUpstreamDrainPolicies(ctx) + if err != nil { + return UpstreamDrainResult{}, err + } + if err := reaper.store.ReplaceUpstreamDrainPolicies(ctx, policies); err != nil { + return UpstreamDrainResult{}, err + } + result := UpstreamDrainResult{} + remaining := reaper.options.BatchSize + utcNow := now.UTC() + for _, policy := range policies { + if policy.Enabled || remaining == 0 { + continue + } + candidates, err := reaper.store.ListDisabledUpstreamDrainCandidates(ctx, utcNow, policy, remaining) + if err != nil { + return result, err + } + if len(candidates) > remaining { + return result, ErrInvalidUpstreamDrainReaper + } + for _, candidate := range candidates { + started, err := reaper.store.BeginDisabledUpstreamDrain(ctx, utcNow, candidate) + if err != nil { + return result, err + } + result.Candidates++ + if started { + result.Started++ + } + } + remaining -= len(candidates) + } + return result, nil +} + +func (reaper *ConfiguredUpstreamDrainReaper) Run(ctx context.Context) error { + if reaper == nil { + return ErrInvalidUpstreamDrainReaper + } + return runScheduler(ctx, reaper.options.PollInterval, func(tickCtx context.Context) (TickResult, error) { + _, err := reaper.Tick(tickCtx) + return TickResult{}, err + }) +} diff --git a/internal/controller/health/upstream_drain_reaper_test.go b/internal/controller/health/upstream_drain_reaper_test.go new file mode 100644 index 0000000..b11a3d2 --- /dev/null +++ b/internal/controller/health/upstream_drain_reaper_test.go @@ -0,0 +1,114 @@ +package health + +import ( + "context" + "testing" + "time" + + "proxy-pool/internal/config" + "proxy-pool/internal/domain/activitypool" + "proxy-pool/internal/domain/adminstate" + proxyDomain "proxy-pool/internal/domain/proxy" +) + +type upstreamDrainPolicySourceStub struct { + policies []activitypool.UpstreamDrainPolicy + err error +} + +func (source upstreamDrainPolicySourceStub) ReadUpstreamDrainPolicies(context.Context) ([]activitypool.UpstreamDrainPolicy, error) { + return append([]activitypool.UpstreamDrainPolicy(nil), source.policies...), source.err +} + +type upstreamDrainConfigurationStub struct { + configuration *config.Config + revision uint64 +} + +type upstreamDrainStateStub struct{ snapshot adminstate.Snapshot } + +func (source upstreamDrainStateStub) Snapshot(context.Context) (adminstate.Snapshot, error) { + return source.snapshot, nil +} + +func (source upstreamDrainConfigurationStub) Snapshot() (*config.Config, uint64) { + return source.configuration, source.revision +} + +func TestConfiguredUpstreamDrainReaperStartsBoundedDisabledAssignments(t *testing.T) { + now := time.Date(2026, 8, 2, 16, 0, 0, 0, time.UTC) + pool := activitypool.NewMemoryPool() + if _, err := pool.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1, + Proxies: []proxyDomain.Proxy{{ + ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080, State: proxyDomain.StateAvailable, + }}, + }); err != nil { + t.Fatalf("UpsertFetched() error = %v", err) + } + assignment, err := pool.Assign(context.Background(), now, "proxy-a", "worker-a", time.Minute) + if err != nil { + t.Fatalf("Assign() error = %v", err) + } + policy := activitypool.UpstreamDrainPolicy{UpstreamID: "provider-a", Revision: 21} + reaper, err := NewConfiguredUpstreamDrainReaper(upstreamDrainPolicySourceStub{policies: []activitypool.UpstreamDrainPolicy{policy}}, pool, + UpstreamDrainReaperOptions{PollInterval: time.Second, BatchSize: 4, Now: func() time.Time { return now }}) + if err != nil { + t.Fatalf("NewConfiguredUpstreamDrainReaper() error = %v", err) + } + result, err := reaper.Tick(context.Background()) + if err != nil || result != (UpstreamDrainResult{Candidates: 1, Started: 1}) { + t.Fatalf("Tick() = %+v, %v", result, err) + } + if current, found, err := pool.Get(context.Background(), "proxy-a"); err != nil || !found || + current.Epoch != assignment.Epoch || !current.Draining { + t.Fatalf("Get(after drain) = %+v, %t, %v", current, found, err) + } +} + +func TestEffectiveUpstreamDrainPolicySourceUsesStableConfigOnlyRevision(t *testing.T) { + configuration := &config.Config{Upstreams: map[string]config.Upstream{ + "provider-b": {Enabled: true}, + "provider-a": {Enabled: false}, + }} + source, err := NewEffectiveUpstreamDrainPolicySource(upstreamDrainConfigurationStub{configuration: configuration}) + if err != nil { + t.Fatalf("NewEffectiveUpstreamDrainPolicySource() error = %v", err) + } + policies, err := source.ReadUpstreamDrainPolicies(context.Background()) + if err != nil || len(policies) != 2 || policies[0] != (activitypool.UpstreamDrainPolicy{ + UpstreamID: "provider-a", Revision: 1, + }) || policies[1] != (activitypool.UpstreamDrainPolicy{ + UpstreamID: "provider-b", Revision: 1, Enabled: true, + }) { + t.Fatalf("ReadUpstreamDrainPolicies() = %+v, %v", policies, err) + } +} + +func TestEffectiveUpstreamDrainPolicySourceCombinesConfigurationAndAdminState(t *testing.T) { + configuration := &config.Config{Upstreams: map[string]config.Upstream{ + "provider-a": {Enabled: false}, + "provider-b": {Enabled: true}, + }} + source, err := NewEffectiveUpstreamDrainPolicySource( + upstreamDrainConfigurationStub{configuration: configuration, revision: 9}, + upstreamDrainStateStub{snapshot: adminstate.Snapshot{ + Config: &adminstate.ConfigRevision{Revision: 9}, + Upstreams: []adminstate.UpstreamState{ + {Name: "provider-a", Enabled: true, Revision: 7}, + {Name: "provider-b", Enabled: false, Revision: 8}, + }, + }}, + ) + if err != nil { + t.Fatalf("NewEffectiveUpstreamDrainPolicySource() error = %v", err) + } + policies, err := source.ReadUpstreamDrainPolicies(context.Background()) + want := []activitypool.UpstreamDrainPolicy{ + {UpstreamID: "provider-a", Revision: 7}, + {UpstreamID: "provider-b", Revision: 8}, + } + if err != nil || len(policies) != len(want) || policies[0] != want[0] || policies[1] != want[1] { + t.Fatalf("ReadUpstreamDrainPolicies() = %+v, %v; want %+v", policies, err, want) + } +} diff --git a/internal/domain/activitypool/pool.go b/internal/domain/activitypool/pool.go index 39263cc..bce4ff6 100644 --- a/internal/domain/activitypool/pool.go +++ b/internal/domain/activitypool/pool.go @@ -162,6 +162,45 @@ type UnhealthyDrainStarter interface { BeginUnhealthyDrain(context.Context, time.Time, UnhealthyDrainCandidate) (bool, error) } +// UpstreamDrainPolicy is the effective, versioned enablement state for one +// Upstream. It is copied into the activity pool so a conditional Drain can +// verify the same policy revision that selected its candidate. +type UpstreamDrainPolicy struct { + UpstreamID string + Revision uint64 + Enabled bool +} + +// UpstreamDrainPolicyWriter replaces the complete effective policy view. A +// complete replacement removes policies for deleted Upstreams as well. +type UpstreamDrainPolicyWriter interface { + ReplaceUpstreamDrainPolicies(context.Context, []UpstreamDrainPolicy) error +} + +// DisabledUpstreamDrainCandidate is a bounded, non-authoritative ownership +// hint for an effectively disabled Upstream. PolicyRevision fences an old +// candidate after the Upstream has been enabled again. +type DisabledUpstreamDrainCandidate struct { + UpstreamID string + PolicyRevision uint64 + ProxyID string + WorkerID string + AssignmentEpoch uint64 +} + +// DisabledUpstreamDrainLister finds candidates from one disabled policy. The +// list is deliberately bounded and must be followed by BeginDisabledUpstreamDrain. +type DisabledUpstreamDrainLister interface { + ListDisabledUpstreamDrainCandidates(context.Context, time.Time, UpstreamDrainPolicy, int) ([]DisabledUpstreamDrainCandidate, error) +} + +// DisabledUpstreamDrainStarter conditionally begins a Drain only while the +// candidate's effective policy revision, ownership and Proxy source still +// agree. False means a normal concurrent change made the candidate stale. +type DisabledUpstreamDrainStarter interface { + BeginDisabledUpstreamDrain(context.Context, time.Time, DisabledUpstreamDrainCandidate) (bool, error) +} + const maximumUnhealthySweepScan = 1024 type Entry struct { @@ -182,6 +221,7 @@ type MemoryPool struct { drains map[string]ownershipDomain.DrainTicket targets map[targetHealthKey]healthDomain.TargetState unhealthy map[string]time.Time + policies map[string]UpstreamDrainPolicy nextEpoch uint64 } @@ -207,6 +247,9 @@ var ( _ Maintainer = (*MemoryPool)(nil) _ UnhealthyRemover = (*MemoryPool)(nil) _ UnhealthyDrainStarter = (*MemoryPool)(nil) + _ UpstreamDrainPolicyWriter = (*MemoryPool)(nil) + _ DisabledUpstreamDrainLister = (*MemoryPool)(nil) + _ DisabledUpstreamDrainStarter = (*MemoryPool)(nil) _ extractionDomain.Store = (*MemoryPool)(nil) _ ownershipDomain.Repository = (*MemoryPool)(nil) _ ownershipDomain.DrainTicketStore = (*MemoryPool)(nil) @@ -221,6 +264,7 @@ func NewMemoryPool() *MemoryPool { drains: make(map[string]ownershipDomain.DrainTicket), targets: make(map[targetHealthKey]healthDomain.TargetState), unhealthy: make(map[string]time.Time), + policies: make(map[string]UpstreamDrainPolicy), } } @@ -708,6 +752,117 @@ func (p *MemoryPool) SweepUnhealthy( return result, nil } +// ReplaceUpstreamDrainPolicies publishes one detached, versioned policy view. +// The pool deliberately owns no configuration data beyond this narrow Drain +// fence, so proxy records remain entirely ephemeral. +func (p *MemoryPool) ReplaceUpstreamDrainPolicies(ctx context.Context, policies []UpstreamDrainPolicy) error { + if ctx == nil { + return ErrInvalidMaintenance + } + if err := ctx.Err(); err != nil { + return err + } + if p == nil || !validUpstreamDrainPolicies(policies) { + return ErrInvalidMaintenance + } + next := make(map[string]UpstreamDrainPolicy, len(policies)) + for _, policy := range policies { + next[policy.UpstreamID] = policy + } + p.mu.Lock() + defer p.mu.Unlock() + if err := ctx.Err(); err != nil { + return err + } + p.policies = next + return nil +} + +// ListDisabledUpstreamDrainCandidates returns at most limit owned AVAILABLE +// proxies for a policy that is still disabled at the supplied revision. +func (p *MemoryPool) ListDisabledUpstreamDrainCandidates( + ctx context.Context, + now time.Time, + policy UpstreamDrainPolicy, + limit int, +) ([]DisabledUpstreamDrainCandidate, error) { + if ctx == nil { + return nil, ErrInvalidMaintenance + } + if err := ctx.Err(); err != nil { + return nil, err + } + if p == nil || now.IsZero() || limit <= 0 || !validUpstreamDrainPolicy(policy) || policy.Enabled { + return nil, ErrInvalidMaintenance + } + p.mu.Lock() + defer p.mu.Unlock() + if err := ctx.Err(); err != nil { + return nil, err + } + p.purgeExpiredLocked(now) + if current, exists := p.policies[policy.UpstreamID]; !exists || current != policy { + return nil, nil + } + proxyIDs := make([]string, 0) + for proxyID, assignment := range p.ownership { + entry, exists := p.entryByIDLocked(proxyID) + if !exists || assignment.Draining || !assignment.ExpiresAt.After(now) || + entry.Proxy.SourceUpstream != policy.UpstreamID || entry.State != proxyDomain.StateAvailable || + entry.OwnerWorkerID != assignment.WorkerID || !entry.UsableUntil.After(now) { + continue + } + proxyIDs = append(proxyIDs, proxyID) + } + sort.Strings(proxyIDs) + if len(proxyIDs) > limit { + proxyIDs = proxyIDs[:limit] + } + result := make([]DisabledUpstreamDrainCandidate, 0, len(proxyIDs)) + for _, proxyID := range proxyIDs { + assignment := p.ownership[proxyID] + result = append(result, DisabledUpstreamDrainCandidate{ + UpstreamID: policy.UpstreamID, PolicyRevision: policy.Revision, ProxyID: proxyID, + WorkerID: assignment.WorkerID, AssignmentEpoch: assignment.Epoch, + }) + } + return result, nil +} + +// BeginDisabledUpstreamDrain is the ownership-side final decision for a +// candidate. Policy, source upstream and assignment are checked under one +// pool lock before the existing Drain ticket flow is entered. +func (p *MemoryPool) BeginDisabledUpstreamDrain( + ctx context.Context, + now time.Time, + candidate DisabledUpstreamDrainCandidate, +) (bool, error) { + if err := ownershipContextError(ctx); err != nil { + return false, err + } + if p == nil || now.IsZero() || !validDisabledUpstreamDrainCandidate(candidate) { + return false, ErrInvalidMaintenance + } + p.mu.Lock() + defer p.mu.Unlock() + if err := ctx.Err(); err != nil { + return false, err + } + p.purgeExpiredLocked(now) + policy, active := p.policies[candidate.UpstreamID] + assignment, owned := p.ownership[candidate.ProxyID] + entry, exists := p.entryByIDLocked(candidate.ProxyID) + if !active || policy.Enabled || policy.Revision != candidate.PolicyRevision || + !owned || !exists || assignment.Draining || !assignment.ExpiresAt.After(now) || + assignment.WorkerID != candidate.WorkerID || assignment.Epoch != candidate.AssignmentEpoch || + entry.Proxy.SourceUpstream != candidate.UpstreamID || entry.State != proxyDomain.StateAvailable || + entry.OwnerWorkerID != candidate.WorkerID { + return false, nil + } + _, started := p.beginDrainLocked(assignment) + return started, nil +} + func (p *MemoryPool) Snapshot(now time.Time) []Entry { if p == nil { return nil @@ -1192,6 +1347,29 @@ func validUnhealthySweepPolicies(policies map[string]time.Duration) bool { return true } +func validUpstreamDrainPolicies(policies []UpstreamDrainPolicy) bool { + seen := make(map[string]struct{}, len(policies)) + for _, policy := range policies { + if !validUpstreamDrainPolicy(policy) { + return false + } + if _, duplicate := seen[policy.UpstreamID]; duplicate { + return false + } + seen[policy.UpstreamID] = struct{}{} + } + return true +} + +func validUpstreamDrainPolicy(policy UpstreamDrainPolicy) bool { + return policy.UpstreamID != "" && policy.Revision > 0 +} + +func validDisabledUpstreamDrainCandidate(candidate DisabledUpstreamDrainCandidate) bool { + return candidate.UpstreamID != "" && candidate.PolicyRevision > 0 && candidate.ProxyID != "" && + candidate.WorkerID != "" && candidate.AssignmentEpoch > 0 +} + func (p *MemoryPool) entryByIDLocked(proxyID string) (Entry, bool) { key, ok := p.keyByID[proxyID] if !ok { diff --git a/internal/domain/activitypool/pool_test.go b/internal/domain/activitypool/pool_test.go index 0c2e38e..5a84a05 100644 --- a/internal/domain/activitypool/pool_test.go +++ b/internal/domain/activitypool/pool_test.go @@ -456,3 +456,52 @@ func poolWithOneProxy(t *testing.T, now time.Time) *MemoryPool { } return pool } + +func TestMemoryPoolDisabledUpstreamDrainFencesPolicyRevision(t *testing.T) { + now := time.Date(2026, 8, 2, 15, 0, 0, 0, time.UTC) + pool := poolWithOneProxy(t, now) + proxyID := pool.Snapshot(now)[0].Proxy.ID + assignment, err := pool.Assign(context.Background(), now, proxyID, "worker-a", time.Minute) + if err != nil { + t.Fatalf("Assign() error = %v", err) + } + disabled := UpstreamDrainPolicy{UpstreamID: "provider-a", Revision: 11} + if err := pool.ReplaceUpstreamDrainPolicies(context.Background(), []UpstreamDrainPolicy{disabled}); err != nil { + t.Fatalf("ReplaceUpstreamDrainPolicies(disabled) error = %v", err) + } + candidates, err := pool.ListDisabledUpstreamDrainCandidates(context.Background(), now, disabled, 1) + if err != nil || len(candidates) != 1 || candidates[0] != (DisabledUpstreamDrainCandidate{ + UpstreamID: "provider-a", PolicyRevision: 11, ProxyID: proxyID, WorkerID: "worker-a", AssignmentEpoch: assignment.Epoch, + }) { + t.Fatalf("ListDisabledUpstreamDrainCandidates() = %+v, %v", candidates, err) + } + if err := pool.ReplaceUpstreamDrainPolicies(context.Background(), []UpstreamDrainPolicy{{ + UpstreamID: "provider-a", Revision: 12, Enabled: true, + }}); err != nil { + t.Fatalf("ReplaceUpstreamDrainPolicies(enabled) error = %v", err) + } + if started, err := pool.BeginDisabledUpstreamDrain(context.Background(), now, candidates[0]); err != nil || started { + t.Fatalf("BeginDisabledUpstreamDrain(stale policy) = %t, %v", started, err) + } + if _, found, err := pool.Get(context.Background(), proxyID); err != nil || !found { + t.Fatalf("Get(after stale candidate) = found:%t, %v", found, err) + } + + if err := pool.ReplaceUpstreamDrainPolicies(context.Background(), []UpstreamDrainPolicy{{ + UpstreamID: "provider-a", Revision: 13, + }}); err != nil { + t.Fatalf("ReplaceUpstreamDrainPolicies(disabled again) error = %v", err) + } + candidates, err = pool.ListDisabledUpstreamDrainCandidates(context.Background(), now, UpstreamDrainPolicy{ + UpstreamID: "provider-a", Revision: 13, + }, 1) + if err != nil || len(candidates) != 1 { + t.Fatalf("ListDisabledUpstreamDrainCandidates(disabled again) = %+v, %v", candidates, err) + } + if started, err := pool.BeginDisabledUpstreamDrain(context.Background(), now, candidates[0]); err != nil || !started { + t.Fatalf("BeginDisabledUpstreamDrain() = %t, %v", started, err) + } + if assignment, found, err := pool.Get(context.Background(), proxyID); err != nil || !found || !assignment.Draining { + t.Fatalf("Get(after drain) = %+v, found:%t, %v", assignment, found, err) + } +} diff --git a/progress.md b/progress.md index 3ec7b24..b23535a 100644 --- a/progress.md +++ b/progress.md @@ -2,6 +2,13 @@ ## 2026-08-02 +- 已实现上游停用的自动 Drain：有效策略由静态配置与 PostgreSQL Admin 管理态共同计算， + Controller 持续发布到 Redis；每个候选绑定上游 revision，只从 `owned(upstream)` 有界索引 + 读取。Redis 在创建 Drain Ticket 前原子复核策略 revision、Proxy 上游、Worker、assignment + epoch 和租约，重启用或重新分配后的旧候选不会排空当前归属。 +- 新增 Memory/Redis 契约与 Controller Reaper 测试，并修复 Provider Supervisor 只看管理态 + 而可能重新启动静态停用上游的问题。`go test -count=1 -timeout 60s ./...` 和 Redis 集成标签 + 测试已通过。 - 已提交并推送 `2af504f feat: expose gateway outcome metrics`：Gateway 请求路径使用 领域 Metrics Observer，将固定阶段的成功/失败聚合到 `proxy_pool_gateway_outcomes_total{stage,result}`，并暴露本地 Outcome 队列满时的 diff --git a/task_plan.md b/task_plan.md index 6873e33..ba92213 100644 --- a/task_plan.md +++ b/task_plan.md @@ -43,7 +43,8 @@ 15. [进行中] 收敛 Worker 发布生命周期：已完成权威 Proxy/Routing 的持续完整快照 刷新，以及 Drain 发起时的持久化 Ticket、Worker 待绑定索引、ownership epoch 推进和 完整排除 Snapshot 屏障绑定；已在 ACK 与 Runtime 零计数之间完成 Redis 原子自动收尾。 - 后续将配置停用接入同一 `BeginDrain` 编排，并补齐可观测闭环。 + 配置与管理态共同决定的上游停用已接入同一 `BeginDrain` 编排：候选绑定上游 revision， + Redis 原子复核策略、归属与 assignment epoch。后续补齐可观测闭环。 ## 串并行关系