diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index d9b9a74..76a3491 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -278,7 +278,8 @@ Profile Reducer 与任务摘要幂等语义;BASIC/EGRESS 结果经活动池窄 Lua 同一原子边界归并,覆盖首次 CHECKING 失败进入 UNHEALTHY、AVAILABLE/SUSPECT 的 阈值降级、精确重放、冲突拒绝与成功恢复。TARGET Profile 也已在 Memory 和 Redis 中独立 归并,以哈希键保存并随代理 TTL 过期,绝不写入 Proxy 全局状态或选择索引。调度器、 -Checker RPC 与独立进程尚未接入,因此本任务保持未完成。 +Controller 到 Checker 的 RPC、任务调度和独立进程尚未接入;Controller 已新增公用 Reducer +作为 Observation 到原子 Store 的唯一归并边界,因此本任务保持未完成。 ## Task 12: Machine-readable Contracts diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 5a6bfba..3308321 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -59,7 +59,8 @@ Routing payload 已由配置顺序和 Admin 当前状态合成并覆盖 Snapshot Outcome 已实现为 Gateway 本地有界队列、微批确认重试和 Controller 的 session/sequence/ 摘要 Redis 栅栏;原始事件不落 Redis 或 PostgreSQL。Checker 已有全局健康 Reducer 与 Memory/Redis 原子状态提交基础;TARGET Profile 以独立、随代理 TTL 过期的 Redis -记录归并,不改写 Proxy 全局状态。任务调度、Checker RPC 和独立执行进程尚未闭环。Snapshot 签发在 Redis 中原子匹配当前 +记录归并,不改写 Proxy 全局状态。Controller 公用 Reducer 已作为 Observation 的唯一状态 +归并边界;任务调度、Checker RPC 和独立执行进程尚未闭环。Snapshot 签发在 Redis 中原子匹配当前 `session_id`,重注册会清除旧引用,迟到旧 Stream 不会覆盖新 session。Controller 在最近成功下发的 Snapshot `valid_until` 到达时关闭流;Gateway 的公用 `SessionSupervisor` 已实现可恢复错误的有界退避重连。Gateway 会校验并执行 Snapshot diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index ebb4671..f370d27 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 消费待实现 | | HEALTH-002 | 健康调度有 jitter、maxInFlight 和分级频率 | 8679-8736 | 配置校验已完成;有界调度器、抖动和分级频率测试待实现 | -| HEALTH-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | 全局 Reducer 已在 Memory/Redis 活动池原子提交连续失败、精确重放和成功恢复;任务调度与 REMOVE 编排待实现 | +| HEALTH-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | Controller 公用 Reducer 已通过 Memory/Redis 活动池原子提交全局连续失败、精确重放和成功恢复;任务调度与 REMOVE 编排待实现 | | 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/internal/config/effective_check.go b/internal/config/effective_check.go new file mode 100644 index 0000000..b732051 --- /dev/null +++ b/internal/config/effective_check.go @@ -0,0 +1,41 @@ +package config + +// EffectiveCheck overlays one upstream's non-zero check fields on the global +// defaults. A non-nil URL slice is an explicit override, including an empty +// slice that intentionally disables URL-based EGRESS checks for that upstream. +func EffectiveCheck(defaults, override Check) Check { + effective := defaults + if override.Interval != 0 { + effective.Interval = override.Interval + } + if override.Jitter != 0 { + effective.Jitter = override.Jitter + } + if override.MaxInFlight != 0 { + effective.MaxInFlight = override.MaxInFlight + } + if override.Timeout != 0 { + effective.Timeout = override.Timeout + } + if override.MaxAttempts != 0 { + effective.MaxAttempts = override.MaxAttempts + } + if override.MaxConsecutiveFailures != 0 { + effective.MaxConsecutiveFailures = override.MaxConsecutiveFailures + } + if override.URLs != nil { + effective.URLs = cloneCheckURLs(override.URLs) + } else { + effective.URLs = cloneCheckURLs(defaults.URLs) + } + return effective +} + +func cloneCheckURLs(source []string) []string { + if source == nil { + return nil + } + result := make([]string, len(source)) + copy(result, source) + return result +} diff --git a/internal/config/effective_check_test.go b/internal/config/effective_check_test.go new file mode 100644 index 0000000..b318691 --- /dev/null +++ b/internal/config/effective_check_test.go @@ -0,0 +1,61 @@ +package config + +import ( + "strings" + "testing" + "time" +) + +func TestEffectiveCheckOverlaysScalarsAndClonesURLs(t *testing.T) { + defaults := Check{ + Interval: Duration(time.Minute), Jitter: 20, MaxInFlight: 12, Timeout: Duration(5 * time.Second), + MaxAttempts: 3, MaxConsecutiveFailures: 2, URLs: []string{"https://egress.example/check"}, + } + override := Check{Timeout: Duration(8 * time.Second), MaxAttempts: 5} + effective := EffectiveCheck(defaults, override) + if effective.Interval != defaults.Interval || effective.Jitter != defaults.Jitter || + effective.MaxInFlight != defaults.MaxInFlight || effective.Timeout != override.Timeout || + effective.MaxAttempts != override.MaxAttempts || + effective.MaxConsecutiveFailures != defaults.MaxConsecutiveFailures || + len(effective.URLs) != 1 || effective.URLs[0] != defaults.URLs[0] { + t.Fatalf("EffectiveCheck() = %+v", effective) + } + effective.URLs[0] = "https://changed.example/check" + if defaults.URLs[0] == effective.URLs[0] { + t.Fatal("EffectiveCheck() aliases default URLs") + } + + effective = EffectiveCheck(defaults, Check{URLs: []string{}}) + if effective.URLs == nil || len(effective.URLs) != 0 { + t.Fatalf("EffectiveCheck(explicit empty URLs) = %+v", effective) + } +} + +func TestValidateUsesEffectiveCheckAndRejectsUnsafeURLs(t *testing.T) { + cfg := mustLoadValidConfig(t) + defaults := cfg.Upstreams["provider-a"].Check + cfg.Defaults.Check = defaults + upstream := cfg.Upstreams["provider-a"] + upstream.Check = Check{Timeout: Duration(7 * time.Second)} + cfg.Upstreams["provider-a"] = upstream + if err := Validate(cfg); err != nil { + t.Fatalf("Validate(effective check) = %v", err) + } + + for _, invalid := range []string{ + "target.example/check", "ftp://target.example/check", "https://user:pass@target.example/check", + "https://target.example/check#fragment", " https://target.example/check", "https://target.example/check", + } { + cfg := mustLoadValidConfig(t) + upstream := cfg.Upstreams["provider-a"] + if strings.HasPrefix(invalid, "https://target.example/check") && invalid == "https://target.example/check" { + upstream.Check.URLs = []string{invalid, invalid} + } else { + upstream.Check.URLs = []string{invalid} + } + cfg.Upstreams["provider-a"] = upstream + if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "check.urls") { + t.Fatalf("Validate(check URL %q) error = %v, want check.urls", invalid, err) + } + } +} diff --git a/internal/config/validate.go b/internal/config/validate.go index e0a6c30..78aad50 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -62,7 +62,11 @@ func Validate(cfg *Config) error { if upstream.Enabled { enabledUpstreams++ } - if err := validateUpstream(name, upstream); err != nil { + effective := upstream + if upstream.Enabled { + effective.Check = EffectiveCheck(cfg.Defaults.Check, upstream.Check) + } + if err := validateUpstream(name, effective); err != nil { return err } } @@ -577,6 +581,22 @@ func validateCheck(scope string, check Check) error { if err := requirePositive(scope+".maxConsecutiveFailures", check.MaxConsecutiveFailures); err != nil { return err } + seenURLs := make(map[string]struct{}, len(check.URLs)) + for index, rawURL := range check.URLs { + if rawURL == "" || strings.TrimSpace(rawURL) != rawURL { + return fmt.Errorf("validate %s.urls[%d]: must be an absolute http or https URL", scope, index) + } + parsed, err := url.Parse(rawURL) + if err != nil || parsed.Scheme == "" || parsed.Host == "" || parsed.User != nil || parsed.Fragment != "" || + (parsed.Scheme != "http" && parsed.Scheme != "https") { + return fmt.Errorf("validate %s.urls[%d]: must be an absolute http or https URL", scope, index) + } + canonical := parsed.String() + if _, exists := seenURLs[canonical]; exists { + return fmt.Errorf("validate %s.urls[%d]: duplicate URL", scope, index) + } + seenURLs[canonical] = struct{}{} + } return nil } diff --git a/internal/controller/health/reducer.go b/internal/controller/health/reducer.go new file mode 100644 index 0000000..b947664 --- /dev/null +++ b/internal/controller/health/reducer.go @@ -0,0 +1,103 @@ +// Package health owns Controller-side reduction of Checker facts. It never +// probes, schedules, or persists raw observations. +package health + +import ( + "context" + "errors" + "reflect" + + "proxy-pool/internal/domain/activitypool" + healthDomain "proxy-pool/internal/domain/health" +) + +var ( + ErrInvalidReducer = errors.New("invalid health reducer") + ErrInvalidThreshold = errors.New("invalid health failure threshold") +) + +// FailureThresholdResolver selects the configured consecutive-failure limit +// for one normalized fact. It keeps policy lookup outside the Checker and +// lets a Controller use different limits for different upstreams. +type FailureThresholdResolver func(context.Context, healthDomain.Observation) (int, error) + +// Result contains only the reduced state written by the corresponding store. +// Exactly one of Global or Target is non-nil for an accepted observation. +type Result struct { + Global *activitypool.Entry + Target *healthDomain.TargetState +} + +// Reducer is the Controller's narrow authority boundary for checker facts. +// The stores commit their own atomic idempotency and ordering semantics. +type Reducer struct { + global activitypool.GlobalHealthStore + target activitypool.TargetHealthStore + threshold FailureThresholdResolver +} + +func NewReducer( + global activitypool.GlobalHealthStore, + target activitypool.TargetHealthStore, + threshold FailureThresholdResolver, +) (*Reducer, error) { + if nilInterface(global) || nilInterface(target) || threshold == nil { + return nil, ErrInvalidReducer + } + return &Reducer{global: global, target: target, threshold: threshold}, nil +} + +func nilInterface(value any) bool { + if value == nil { + return true + } + reflected := reflect.ValueOf(value) + switch reflected.Kind() { + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: + return reflected.IsNil() + default: + return false + } +} + +// Apply normalizes one immutable Checker fact, resolves its configured policy, +// and commits it through exactly one atomic activity-pool store method. +func (r *Reducer) Apply(ctx context.Context, observation healthDomain.Observation) (Result, error) { + if ctx == nil || r == nil || r.global == nil || r.target == nil || r.threshold == nil { + return Result{}, ErrInvalidReducer + } + if err := ctx.Err(); err != nil { + return Result{}, err + } + normalized, err := healthDomain.NormalizeObservation(observation) + if err != nil { + return Result{}, err + } + maxConsecutiveFailures, err := r.threshold(ctx, normalized) + if err != nil { + return Result{}, err + } + if maxConsecutiveFailures <= 0 { + return Result{}, ErrInvalidThreshold + } + switch normalized.Level { + case healthDomain.LevelBasic, healthDomain.LevelEgress: + entry, err := r.global.ApplyGlobalObservation(ctx, activitypool.GlobalHealthCommand{ + Observation: normalized, MaxConsecutiveFailures: maxConsecutiveFailures, + }) + if err != nil { + return Result{}, err + } + return Result{Global: &entry}, nil + case healthDomain.LevelTarget: + state, err := r.target.ApplyTargetObservation(ctx, activitypool.TargetHealthCommand{ + Observation: normalized, MaxConsecutiveFailures: maxConsecutiveFailures, + }) + if err != nil { + return Result{}, err + } + return Result{Target: &state}, nil + default: + return Result{}, healthDomain.ErrInvalidObservation + } +} diff --git a/internal/controller/health/reducer_test.go b/internal/controller/health/reducer_test.go new file mode 100644 index 0000000..4ac54d3 --- /dev/null +++ b/internal/controller/health/reducer_test.go @@ -0,0 +1,96 @@ +package health + +import ( + "context" + "errors" + "testing" + "time" + + "proxy-pool/internal/domain/activitypool" + healthDomain "proxy-pool/internal/domain/health" +) + +func TestReducerRoutesObservationsToTheirIsolatedStores(t *testing.T) { + global := &recordingGlobalStore{} + target := &recordingTargetStore{} + var policyFacts []healthDomain.Observation + reducer, err := NewReducer(global, target, func(_ context.Context, observation healthDomain.Observation) (int, error) { + policyFacts = append(policyFacts, observation) + return 3, nil + }) + if err != nil { + t.Fatalf("NewReducer(): %v", err) + } + now := time.Date(2026, 7, 31, 13, 0, 0, 0, time.UTC) + globalResult, err := reducer.Apply(context.Background(), reducerObservation(healthDomain.LevelBasic, now)) + if err != nil || globalResult.Global == nil || globalResult.Target != nil || len(global.commands) != 1 || len(target.commands) != 0 { + t.Fatalf("Apply(BASIC) = %+v, %v; global=%d target=%d", globalResult, err, len(global.commands), len(target.commands)) + } + if global.commands[0].MaxConsecutiveFailures != 3 || global.commands[0].Observation.Level != healthDomain.LevelBasic { + t.Fatalf("global command = %+v", global.commands[0]) + } + + targetResult, err := reducer.Apply(context.Background(), reducerObservation(healthDomain.LevelTarget, now.Add(time.Second))) + if err != nil || targetResult.Global != nil || targetResult.Target == nil || len(global.commands) != 1 || len(target.commands) != 1 { + t.Fatalf("Apply(TARGET) = %+v, %v; global=%d target=%d", targetResult, err, len(global.commands), len(target.commands)) + } + if target.commands[0].MaxConsecutiveFailures != 3 || target.commands[0].Observation.Level != healthDomain.LevelTarget { + t.Fatalf("target command = %+v", target.commands[0]) + } + if len(policyFacts) != 2 || policyFacts[0].ObservedAt.Location() != time.UTC || policyFacts[1].RoutingName != "route-a" { + t.Fatalf("policy observations = %+v", policyFacts) + } +} + +func TestReducerRejectsInvalidPolicyAndDoesNotCallStores(t *testing.T) { + global := &recordingGlobalStore{} + target := &recordingTargetStore{} + reducer, err := NewReducer(global, target, func(context.Context, healthDomain.Observation) (int, error) { return 0, nil }) + if err != nil { + t.Fatalf("NewReducer(): %v", err) + } + if _, err := reducer.Apply(context.Background(), reducerObservation(healthDomain.LevelEgress, time.Now())); !errors.Is(err, ErrInvalidThreshold) { + t.Fatalf("Apply(zero threshold) error = %v, want ErrInvalidThreshold", err) + } + if len(global.commands) != 0 || len(target.commands) != 0 { + t.Fatalf("stores called with invalid policy: global=%d target=%d", len(global.commands), len(target.commands)) + } + + if reducer, err := NewReducer(nil, target, func(context.Context, healthDomain.Observation) (int, error) { return 1, nil }); err == nil || reducer != nil { + t.Fatalf("NewReducer(nil global) = (%v, %v)", reducer, err) + } + var typedNilGlobal *recordingGlobalStore + if reducer, err := NewReducer(typedNilGlobal, target, func(context.Context, healthDomain.Observation) (int, error) { return 1, nil }); err == nil || reducer != nil { + t.Fatalf("NewReducer(typed nil global) = (%v, %v)", reducer, err) + } +} + +func reducerObservation(level healthDomain.Level, observedAt time.Time) healthDomain.Observation { + observation := healthDomain.Observation{ + TaskID: "task-a", ProxyID: "proxy-a", Level: level, Success: true, + Latency: time.Millisecond, ObservedAt: observedAt, + } + if level == healthDomain.LevelTarget { + observation.RoutingName = "route-a" + observation.TargetURL = "https://target.example/check" + } + return observation +} + +type recordingGlobalStore struct { + commands []activitypool.GlobalHealthCommand +} + +func (store *recordingGlobalStore) ApplyGlobalObservation(_ context.Context, command activitypool.GlobalHealthCommand) (activitypool.Entry, error) { + store.commands = append(store.commands, command) + return activitypool.Entry{}, nil +} + +type recordingTargetStore struct { + commands []activitypool.TargetHealthCommand +} + +func (store *recordingTargetStore) ApplyTargetObservation(_ context.Context, command activitypool.TargetHealthCommand) (healthDomain.TargetState, error) { + store.commands = append(store.commands, command) + return healthDomain.TargetState{Status: healthDomain.TargetAvailable}, nil +}