From 0a904a7b6bf8672a86efbcec2987ef9fc48a2042 Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 31 Jul 2026 17:55:06 +0800 Subject: [PATCH] feat: add atomic global health reducer --- docs/development/implementation-plan.md | 6 + docs/requirements/completion-audit.md | 4 +- docs/requirements/traceability.md | 4 +- internal/adapters/redisactivity/codec.go | 53 +-- .../global_health_integration_test.go | 74 ++++ internal/adapters/redisactivity/health.go | 71 ++++ .../adapters/redisactivity/scripts/health.lua | 64 ++- .../adapters/redisactivity/scripts/upsert.lua | 4 + internal/adapters/redisactivity/upsert.go | 13 +- .../domain/activitypool/global_health_test.go | 102 +++++ internal/domain/activitypool/pool.go | 89 ++++- internal/domain/health/health.go | 364 ++++++++++++++++++ internal/domain/health/health_test.go | 95 +++++ 13 files changed, 908 insertions(+), 35 deletions(-) create mode 100644 internal/adapters/redisactivity/global_health_integration_test.go create mode 100644 internal/domain/activitypool/global_health_test.go create mode 100644 internal/domain/health/health.go create mode 100644 internal/domain/health/health_test.go diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index bc3ddb9..2f78594 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -273,6 +273,12 @@ Outcome 上报已完成为有界队列、批次序列/摘要栅栏和确认重 - [ ] Ensure target failures affect only the target profile. - [ ] Add fixture target server and deterministic clock/scheduler tests. +当前进度（2026-07-31）：已新增不依赖 gRPC/Redis 的 `domain/health` Observation、 +全局/Target Profile Reducer 与任务摘要幂等语义；BASIC/EGRESS 结果经活动池窄接口在 +Memory 和 Redis Lua 同一原子边界归并，覆盖首次 CHECKING 失败进入 UNHEALTHY、 +AVAILABLE/SUSPECT 的阈值降级、精确重放、冲突拒绝与成功恢复。TARGET 仍只在纯领域 +Profile 中归并，尚未接入持久化、调度器、Checker RPC 或独立进程，因此本任务保持未完成。 + ## Task 12: Machine-readable Contracts **Files:** `api/openapi/proxy-pool.yaml`, `api/proto/controlplane/v1/controlplane.proto`, diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 0cf5672..6bde411 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -57,7 +57,9 @@ Routing 一起纳入 Snapshot checksum，仅在 mTLS 和 Gateway 当前内存 View 中处理。Gateway Routing payload 已由配置顺序和 Admin 当前状态合成并覆盖 Snapshot checksum，其运行时消费、 Outcome 已实现为 Gateway 本地有界队列、微批确认重试和 Controller 的 session/sequence/ -摘要 Redis 栅栏；原始事件不落 Redis 或 PostgreSQL。Checker 尚未闭环。Snapshot 签发在 Redis 中原子匹配当前 +摘要 Redis 栅栏；原始事件不落 Redis 或 PostgreSQL。Checker 已有全局健康 Reducer +与 Memory/Redis 原子状态提交基础，但任务调度、Checker RPC、TARGET Profile 持久化和 +独立执行进程尚未闭环。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 a9d7576..88fa4b3 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -79,9 +79,9 @@ | ID | 最终需求 | 来源 | 验证证据 | |---|---|---|---| -| HEALTH-001 | 全局健康与 Routing/目标健康分离 | 221-270, 8679-8708 | Protobuf/配置契约已定义；目标 Profile 与健康 Reducer 待实现 | +| HEALTH-001 | 全局健康与 Routing/目标健康分离 | 221-270, 8679-8708 | `domain/health` 已将 BASIC/EGRESS 全局 Reducer 与 TARGET Profile Reducer 分离；TARGET 持久化和 Routing 消费待实现 | | HEALTH-002 | 健康调度有 jitter、maxInFlight 和分级频率 | 8679-8736 | 配置校验已完成；有界调度器、抖动和分级频率测试待实现 | -| HEALTH-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | Proxy 状态迁移骨架已完成；连续失败 Reducer 与活动池原子提交待实现 | +| HEALTH-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | 全局 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/adapters/redisactivity/codec.go b/internal/adapters/redisactivity/codec.go index 0ba5e4c..78408dc 100644 --- a/internal/adapters/redisactivity/codec.go +++ b/internal/adapters/redisactivity/codec.go @@ -18,27 +18,31 @@ var ErrInvalidRecord = errors.New("invalid redis activity record") const recordVersion = 1 type proxyRecord struct { - Version int `json:"version"` - ID string `json:"id"` - Scheme string `json:"scheme"` - Host string `json:"host"` - Port int64 `json:"port"` - Username string `json:"username,omitempty"` - Password string `json:"password,omitempty"` - CredentialVersion string `json:"credentialVersion,omitempty"` - SourceUpstream string `json:"sourceUpstream"` - CreatedAtMS int64 `json:"createdAtMs"` - ExpiresAtMS int64 `json:"expiresAtMs"` - UsableUntilMS int64 `json:"usableUntilMs"` - LastCheckedAtMS int64 `json:"lastCheckedAtMs,omitempty"` - LastSuccessAtMS int64 `json:"lastSuccessAtMs,omitempty"` - LatencyNS int64 `json:"latencyNs"` - MaxConcurrency int64 `json:"maxConcurrency"` - State string `json:"state"` - Tags map[string]string `json:"tags,omitempty"` - OwnerWorkerID string `json:"ownerWorkerId,omitempty"` - OwnerIndexKey string `json:"ownerIndexKey"` - IndexKeys []string `json:"indexKeys,omitempty"` + Version int `json:"version"` + ID string `json:"id"` + Scheme string `json:"scheme"` + Host string `json:"host"` + Port int64 `json:"port"` + Username string `json:"username,omitempty"` + Password string `json:"password,omitempty"` + CredentialVersion string `json:"credentialVersion,omitempty"` + SourceUpstream string `json:"sourceUpstream"` + CreatedAtMS int64 `json:"createdAtMs"` + ExpiresAtMS int64 `json:"expiresAtMs"` + UsableUntilMS int64 `json:"usableUntilMs"` + LastCheckedAtMS int64 `json:"lastCheckedAtMs,omitempty"` + LastSuccessAtMS int64 `json:"lastSuccessAtMs,omitempty"` + ConsecutiveFailures int `json:"consecutiveFailures,omitempty"` + LastHealthTaskID string `json:"lastHealthTaskId,omitempty"` + LastHealthDigest string `json:"lastHealthDigest,omitempty"` + LastHealthObservedAtMS int64 `json:"lastHealthObservedAtMs,omitempty"` + LatencyNS int64 `json:"latencyNs"` + MaxConcurrency int64 `json:"maxConcurrency"` + State string `json:"state"` + Tags map[string]string `json:"tags,omitempty"` + OwnerWorkerID string `json:"ownerWorkerId,omitempty"` + OwnerIndexKey string `json:"ownerIndexKey"` + IndexKeys []string `json:"indexKeys,omitempty"` } type ownershipRecord struct { @@ -204,11 +208,16 @@ func validateProxyRecord(record proxyRecord) error { record.Port <= 0 || record.Port > 65_535 || record.SourceUpstream == "" || record.CreatedAtMS <= 0 || record.ExpiresAtMS <= 0 || record.UsableUntilMS <= 0 || record.UsableUntilMS > record.ExpiresAtMS || record.LastCheckedAtMS < 0 || - record.LastSuccessAtMS < 0 || record.LatencyNS < 0 || record.MaxConcurrency < 0 || + record.LastSuccessAtMS < 0 || record.ConsecutiveFailures < 0 || record.LatencyNS < 0 || record.MaxConcurrency < 0 || !validScheme(record.Scheme) || !validProxyState(record.State) || record.OwnerIndexKey == "" || !strings.Contains(record.OwnerIndexKey, "{activity}") { return ErrInvalidRecord } + if (record.LastHealthTaskID == "") != (record.LastHealthDigest == "") || + (record.LastHealthDigest != "" && (!validDigest(record.LastHealthDigest) || record.LastHealthObservedAtMS <= 0)) || + record.LastHealthObservedAtMS < 0 { + return ErrInvalidRecord + } for _, key := range record.IndexKeys { if key == "" || !strings.Contains(key, "{activity}") { return ErrInvalidRecord diff --git a/internal/adapters/redisactivity/global_health_integration_test.go b/internal/adapters/redisactivity/global_health_integration_test.go new file mode 100644 index 0000000..a119e15 --- /dev/null +++ b/internal/adapters/redisactivity/global_health_integration_test.go @@ -0,0 +1,74 @@ +//go:build integration + +package redisactivity + +import ( + "context" + "errors" + "testing" + "time" + + "proxy-pool/internal/domain/activitypool" + healthDomain "proxy-pool/internal/domain/health" + proxyDomain "proxy-pool/internal/domain/proxy" +) + +func TestRedisAppliesGlobalHealthObservationAtomically(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) + } + first := redisGlobalObservation("task-1", false, now.Add(2*time.Second)) + failed, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{ + Observation: first, MaxConsecutiveFailures: 3, + }) + if err != nil || failed.State != proxyDomain.StateUnhealthy || failed.GlobalHealth.ConsecutiveFailures != 1 { + t.Fatalf("ApplyGlobalObservation(initial failure) = %+v, %v", failed, err) + } + replayed, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{ + Observation: first, MaxConsecutiveFailures: 3, + }) + if err != nil || replayed.GlobalHealth != failed.GlobalHealth { + t.Fatalf("ApplyGlobalObservation(replay) = %+v, %v", replayed, err) + } + conflicting := first + conflicting.Success = true + conflicting.FailureClass = "" + if _, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{ + Observation: conflicting, MaxConsecutiveFailures: 3, + }); !errors.Is(err, healthDomain.ErrConflictingObservation) { + t.Fatalf("ApplyGlobalObservation(conflict) error = %v, want ErrConflictingObservation", err) + } + if _, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: now.Add(3 * time.Second), NextState: proxyDomain.StateChecking, + }); err != nil { + t.Fatalf("ApplyHealth(recheck): %v", err) + } + recovered, err := fixture.Adapter.ApplyGlobalObservation(context.Background(), activitypool.GlobalHealthCommand{ + Observation: redisGlobalObservation("task-2", true, now.Add(4*time.Second)), MaxConsecutiveFailures: 3, + }) + if err != nil || recovered.State != proxyDomain.StateAvailable || recovered.GlobalHealth.ConsecutiveFailures != 0 || + recovered.GlobalHealth.LastObservedAt != now.Add(4*time.Second) { + t.Fatalf("ApplyGlobalObservation(recovery) = %+v, %v", recovered, err) + } +} + +func redisGlobalObservation(taskID string, success bool, observedAt time.Time) healthDomain.Observation { + observation := healthDomain.Observation{ + TaskID: taskID, ProxyID: "proxy-a", Level: healthDomain.LevelBasic, + Success: success, Latency: 10 * time.Millisecond, ObservedAt: observedAt, + } + if !success { + observation.FailureClass = "timeout" + } + return observation +} diff --git a/internal/adapters/redisactivity/health.go b/internal/adapters/redisactivity/health.go index 21f0be9..f04dade 100644 --- a/internal/adapters/redisactivity/health.go +++ b/internal/adapters/redisactivity/health.go @@ -2,11 +2,14 @@ package redisactivity import ( "context" + "encoding/hex" "proxy-pool/internal/domain/activitypool" + healthDomain "proxy-pool/internal/domain/health" ) var _ activitypool.HealthStore = (*Adapter)(nil) +var _ activitypool.GlobalHealthStore = (*Adapter)(nil) func (a *Adapter) ApplyHealth(ctx context.Context, update activitypool.HealthUpdate) (activitypool.Entry, error) { if ctx == nil { @@ -56,3 +59,71 @@ func (a *Adapter) ApplyHealth(ctx context.Context, update activitypool.HealthUpd return activitypool.Entry{}, invalidScriptReply("unexpected health status") } } + +// ApplyGlobalObservation reduces one BASIC or EGRESS observation inside the +// same Redis Lua transaction that updates state indexes and inventory. +func (a *Adapter) ApplyGlobalObservation( + ctx context.Context, + command activitypool.GlobalHealthCommand, +) (activitypool.Entry, error) { + if ctx == nil { + return activitypool.Entry{}, activitypool.ErrInvalidHealthUpdate + } + if err := ctx.Err(); err != nil { + return activitypool.Entry{}, err + } + normalized, err := healthDomain.NormalizeObservation(command.Observation) + if err != nil { + return activitypool.Entry{}, err + } + if a == nil || normalized.Level == healthDomain.LevelTarget || command.MaxConsecutiveFailures <= 0 { + return activitypool.Entry{}, activitypool.ErrInvalidHealthUpdate + } + digest, err := healthDomain.ObservationDigest(normalized) + if err != nil { + return activitypool.Entry{}, err + } + operationID, err := newOperationID() + if err != nil { + return activitypool.Entry{}, err + } + success := "0" + if normalized.Success { + success = "1" + } + result, err := runScript(ctx, a.client, healthScript, []string{ + a.keys.records, a.keys.unique, a.keys.idkeys, a.keys.expiry, a.keys.available, + a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry, + a.keys.operation(operationID), + }, normalized.ObservedAt.UnixMilli(), "", int64(normalized.Latency), + a.options.CleanupLimit, operationTTLMillis(a.options.OperationTTL), normalized.ProxyID, + "global", success, command.MaxConsecutiveFailures, normalized.TaskID, hex.EncodeToString(digest[:])) + if err != nil { + return activitypool.Entry{}, err + } + var reply healthScriptReply + if err := decodeScriptResult(result, &reply); err != nil { + return activitypool.Entry{}, err + } + switch reply.Status { + case scriptNotFound: + return activitypool.Entry{}, activitypool.ErrActivityNotFound + case scriptStale: + return activitypool.Entry{}, healthDomain.ErrStaleObservation + case scriptConflict: + return activitypool.Entry{}, healthDomain.ErrConflictingObservation + case scriptInvalid: + return activitypool.Entry{}, activitypool.ErrInvalidHealthUpdate + case scriptOK: + if reply.Record == "" { + return activitypool.Entry{}, invalidScriptReply("global health reply omitted record") + } + record, err := decodeProxyRecord(reply.Record) + if err != nil { + return activitypool.Entry{}, invalidScriptReply("global health reply contained an invalid record") + } + return proxyRecordEntry(record), nil + default: + return activitypool.Entry{}, invalidScriptReply("unexpected global health status") + } +} diff --git a/internal/adapters/redisactivity/scripts/health.lua b/internal/adapters/redisactivity/scripts/health.lua index f3c6ed7..6c876ca 100644 --- a/internal/adapters/redisactivity/scripts/health.lua +++ b/internal/adapters/redisactivity/scripts/health.lua @@ -15,6 +15,11 @@ local latency_ns = tonumber(ARGV[3]) local cleanup_limit = tonumber(ARGV[4]) local operation_ttl_ms = tonumber(ARGV[5]) local proxy_id = ARGV[6] +local mode = ARGV[7] or 'transition' +local success = ARGV[8] +local max_consecutive_failures = tonumber(ARGV[9]) +local task_id = ARGV[10] +local observation_digest = ARGV[11] local committed = redis.call('GET', operation_key) if committed then @@ -194,14 +199,61 @@ if tonumber(record.expiresAtMs) <= checked_at_ms then end local last_checked_at_ms = tonumber(record.lastCheckedAtMs or '0') -if checked_at_ms < last_checked_at_ms then - return finish({status = 'stale'}) -end -if checked_at_ms == last_checked_at_ms then - if record.state ~= next_state then +if mode == 'global' then + if (success ~= '0' and success ~= '1') or not max_consecutive_failures or + max_consecutive_failures <= 0 or type(task_id) ~= 'string' or task_id == '' or + type(observation_digest) ~= 'string' or string.len(observation_digest) ~= 64 then + return finish({status = 'invalid'}) + end + local last_task_id = record.lastHealthTaskId or '' + local last_observation_digest = record.lastHealthDigest or '' + if last_task_id == task_id then + if last_observation_digest == observation_digest then + return finish({status = 'ok', record = raw}, tonumber(record.expiresAtMs)) + end + return finish({status = 'conflict'}) + end + if checked_at_ms <= last_checked_at_ms then return finish({status = 'stale'}) end - return finish({status = 'ok', record = raw}, tonumber(record.expiresAtMs)) + local consecutive_failures = tonumber(record.consecutiveFailures or '0') + if not consecutive_failures or consecutive_failures < 0 then + return finish({status = 'invalid'}) + end + if success == '1' then + if record.state ~= 'CHECKING' and record.state ~= 'AVAILABLE' and record.state ~= 'SUSPECT' then + return finish({status = 'invalid'}) + end + next_state = 'AVAILABLE' + consecutive_failures = 0 + else + consecutive_failures = consecutive_failures + 1 + if record.state == 'CHECKING' then + next_state = 'UNHEALTHY' + elseif record.state == 'AVAILABLE' or record.state == 'SUSPECT' then + if consecutive_failures >= max_consecutive_failures then + next_state = 'UNHEALTHY' + else + next_state = 'SUSPECT' + end + else + return finish({status = 'invalid'}) + end + end + record.consecutiveFailures = consecutive_failures + record.lastHealthTaskId = task_id + record.lastHealthDigest = observation_digest + record.lastHealthObservedAtMs = checked_at_ms +else + if checked_at_ms < last_checked_at_ms then + return finish({status = 'stale'}) + end + if checked_at_ms == last_checked_at_ms then + if record.state ~= next_state then + return finish({status = 'stale'}) + end + return finish({status = 'ok', record = raw}, tonumber(record.expiresAtMs)) + end end if record.state ~= next_state and not (transitions[record.state] and transitions[record.state][next_state]) then return finish({status = 'invalid'}) diff --git a/internal/adapters/redisactivity/scripts/upsert.lua b/internal/adapters/redisactivity/scripts/upsert.lua index bd8283a..1b43dc0 100644 --- a/internal/adapters/redisactivity/scripts/upsert.lua +++ b/internal/adapters/redisactivity/scripts/upsert.lua @@ -227,6 +227,10 @@ for _, candidate in ipairs(candidates) do incoming.state = current.state incoming.lastCheckedAtMs = current.lastCheckedAtMs incoming.lastSuccessAtMs = current.lastSuccessAtMs + incoming.consecutiveFailures = current.consecutiveFailures + incoming.lastHealthTaskId = current.lastHealthTaskId + incoming.lastHealthDigest = current.lastHealthDigest + incoming.lastHealthObservedAtMs = current.lastHealthObservedAtMs incoming.latencyNs = current.latencyNs incoming.ownerWorkerId = current.ownerWorkerId local encoded = cjson.encode(incoming) diff --git a/internal/adapters/redisactivity/upsert.go b/internal/adapters/redisactivity/upsert.go index 9f43618..31cf9f8 100644 --- a/internal/adapters/redisactivity/upsert.go +++ b/internal/adapters/redisactivity/upsert.go @@ -11,6 +11,7 @@ import ( "proxy-pool/internal/domain/activitypool" extractionDomain "proxy-pool/internal/domain/extraction" + healthDomain "proxy-pool/internal/domain/health" proxyDomain "proxy-pool/internal/domain/proxy" "proxy-pool/internal/platform/credentials" ) @@ -252,9 +253,19 @@ func proxyRecordEntry(record proxyRecord) activitypool.Entry { value := time.UnixMilli(record.LastSuccessAtMS).UTC() proxy.LastSuccessAt = &value } + global := healthDomain.GlobalState{State: proxyDomain.State(record.State)} + if record.LastHealthTaskID != "" { + digest, err := hex.DecodeString(record.LastHealthDigest) + if err == nil && len(digest) == sha256.Size && record.LastHealthObservedAtMS > 0 { + copy(global.LastObservationDigest[:], digest) + global.LastTaskID = record.LastHealthTaskID + global.LastObservedAt = time.UnixMilli(record.LastHealthObservedAtMS).UTC() + global.ConsecutiveFailures = record.ConsecutiveFailures + } + } return activitypool.Entry{ Proxy: proxy, UsableUntil: usableUntil, - OwnerWorkerID: record.OwnerWorkerID, State: proxyDomain.State(record.State), + OwnerWorkerID: record.OwnerWorkerID, State: proxyDomain.State(record.State), GlobalHealth: global, } } diff --git a/internal/domain/activitypool/global_health_test.go b/internal/domain/activitypool/global_health_test.go new file mode 100644 index 0000000..aaa7972 --- /dev/null +++ b/internal/domain/activitypool/global_health_test.go @@ -0,0 +1,102 @@ +package activitypool + +import ( + "context" + "errors" + "testing" + "time" + + healthDomain "proxy-pool/internal/domain/health" + proxyDomain "proxy-pool/internal/domain/proxy" +) + +func TestMemoryPoolAppliesGlobalObservationAtomically(t *testing.T) { + now := time.Date(2026, 7, 31, 11, 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) + } + failed, err := pool.ApplyGlobalObservation(context.Background(), GlobalHealthCommand{ + Observation: healthObservation("task-1", false, now.Add(2*time.Second)), MaxConsecutiveFailures: 3, + }) + if err != nil || failed.State != proxyDomain.StateUnhealthy || failed.GlobalHealth.ConsecutiveFailures != 1 { + t.Fatalf("ApplyGlobalObservation(initial failure) = %+v, %v", failed, err) + } + if _, err := pool.ApplyHealth(context.Background(), HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: now.Add(3 * time.Second), NextState: proxyDomain.StateChecking, + }); err != nil { + t.Fatalf("ApplyHealth(recheck): %v", err) + } + available, err := pool.ApplyGlobalObservation(context.Background(), GlobalHealthCommand{ + Observation: healthObservation("task-2", true, now.Add(4*time.Second)), MaxConsecutiveFailures: 3, + }) + if err != nil || available.State != proxyDomain.StateAvailable || available.GlobalHealth.ConsecutiveFailures != 0 || + available.Proxy.LastSuccessAt == nil || !available.Proxy.LastSuccessAt.Equal(now.Add(4*time.Second)) { + t.Fatalf("ApplyGlobalObservation(recovery) = %+v, %v", available, err) + } +} + +func TestMemoryPoolGlobalObservationReplayDoesNotIncreaseFailureStreak(t *testing.T) { + now := time.Date(2026, 7, 31, 11, 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) + } + command := GlobalHealthCommand{Observation: healthObservation("task-1", false, now.Add(2*time.Second)), MaxConsecutiveFailures: 3} + first, err := pool.ApplyGlobalObservation(context.Background(), command) + if err != nil { + t.Fatalf("ApplyGlobalObservation(first): %v", err) + } + replayed, err := pool.ApplyGlobalObservation(context.Background(), command) + if err != nil || replayed.GlobalHealth != first.GlobalHealth { + t.Fatalf("ApplyGlobalObservation(replay) = %+v, %v", replayed, err) + } + conflicting := command + conflicting.Observation.Success = true + conflicting.Observation.FailureClass = "" + if _, err := pool.ApplyGlobalObservation(context.Background(), conflicting); !errors.Is(err, healthDomain.ErrConflictingObservation) { + t.Fatalf("ApplyGlobalObservation(conflict) error = %v, want ErrConflictingObservation", err) + } +} + +func TestMemoryPoolRejectsTargetObservationFromGlobalStore(t *testing.T) { + now := time.Date(2026, 7, 31, 11, 0, 0, 0, time.UTC) + pool := seededHealthPool(t, now) + _, err := pool.ApplyGlobalObservation(context.Background(), GlobalHealthCommand{Observation: healthDomain.Observation{ + TaskID: "task-target", ProxyID: "proxy-a", Level: healthDomain.LevelTarget, RoutingName: "route-a", + TargetURL: "https://target.example/", Success: false, FailureClass: "target_403", ObservedAt: now.Add(time.Second), + }, MaxConsecutiveFailures: 2}) + if !errors.Is(err, ErrInvalidHealthUpdate) { + t.Fatalf("ApplyGlobalObservation(target) error = %v, want ErrInvalidHealthUpdate", err) + } +} + +func seededHealthPool(t *testing.T, now time.Time) *MemoryPool { + t.Helper() + pool := NewMemoryPool() + _, err := pool.UpsertFetched(context.Background(), "provider-a", FetchedBatch{ + ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1, + Proxies: []proxyDomain.Proxy{{ + ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080, + }}, + }) + if err != nil { + t.Fatalf("UpsertFetched(): %v", err) + } + return pool +} + +func healthObservation(taskID string, success bool, observedAt time.Time) healthDomain.Observation { + observation := healthDomain.Observation{ + TaskID: taskID, ProxyID: "proxy-a", Level: healthDomain.LevelBasic, + Success: success, Latency: 10 * time.Millisecond, ObservedAt: observedAt, + } + if !success { + observation.FailureClass = "timeout" + } + return observation +} diff --git a/internal/domain/activitypool/pool.go b/internal/domain/activitypool/pool.go index 878b92d..9118071 100644 --- a/internal/domain/activitypool/pool.go +++ b/internal/domain/activitypool/pool.go @@ -10,6 +10,7 @@ import ( "time" extractionDomain "proxy-pool/internal/domain/extraction" + healthDomain "proxy-pool/internal/domain/health" ownershipDomain "proxy-pool/internal/domain/ownership" proxyDomain "proxy-pool/internal/domain/proxy" ) @@ -77,6 +78,18 @@ type HealthStore interface { ApplyHealth(context.Context, HealthUpdate) (Entry, error) } +// GlobalHealthCommand carries a BASIC or EGRESS fact to the authoritative +// activity pool. TARGET observations use a separate profile store and never +// enter this command. +type GlobalHealthCommand struct { + Observation healthDomain.Observation + MaxConsecutiveFailures int +} + +type GlobalHealthStore interface { + ApplyGlobalObservation(context.Context, GlobalHealthCommand) (Entry, error) +} + type InventoryReader interface { Inventory(context.Context, string, time.Time) (Inventory, error) } @@ -94,6 +107,7 @@ type Entry struct { UsableUntil time.Time OwnerWorkerID string State proxyDomain.State + GlobalHealth healthDomain.GlobalState } type MemoryPool struct { @@ -115,6 +129,7 @@ type idempotencyEntry struct { var ( _ Upserter = (*MemoryPool)(nil) _ HealthStore = (*MemoryPool)(nil) + _ GlobalHealthStore = (*MemoryPool)(nil) _ InventoryReader = (*MemoryPool)(nil) _ StateInventoryReader = (*MemoryPool)(nil) _ Maintainer = (*MemoryPool)(nil) @@ -221,6 +236,7 @@ func (p *MemoryPool) UpsertFetched(ctx context.Context, upstreamID string, batch UsableUntil: usableUntil, OwnerWorkerID: current.OwnerWorkerID, State: current.State, + GlobalHealth: current.GlobalHealth, } continue } @@ -233,9 +249,10 @@ func (p *MemoryPool) UpsertFetched(ctx context.Context, upstreamID string, batch candidate.ID = stableProxyID(key) } p.entries[key] = Entry{ - Proxy: cloneProxy(candidate), - UsableUntil: usableUntil, - State: candidate.State, + Proxy: cloneProxy(candidate), + UsableUntil: usableUntil, + State: candidate.State, + GlobalHealth: healthDomain.GlobalState{State: candidate.State}, } p.keyByID[candidate.ID] = key if managedActivityState(candidate.State) { @@ -284,6 +301,7 @@ func (p *MemoryPool) ApplyHealth(ctx context.Context, update HealthUpdate) (Entr checkedAt := update.CheckedAt.UTC() entry.State = update.NextState entry.Proxy.State = update.NextState + entry.GlobalHealth.State = update.NextState entry.Proxy.LastCheckedAt = &checkedAt entry.Proxy.Latency = update.Latency if update.NextState == proxyDomain.StateAvailable { @@ -295,6 +313,65 @@ func (p *MemoryPool) ApplyHealth(ctx context.Context, update HealthUpdate) (Entr return entry, nil } +// ApplyGlobalObservation atomically reduces a BASIC or EGRESS Checker fact +// against the current Proxy health state and publishes the resulting state. +func (p *MemoryPool) ApplyGlobalObservation( + ctx context.Context, + command GlobalHealthCommand, +) (Entry, error) { + if ctx == nil { + return Entry{}, ErrInvalidHealthUpdate + } + if err := ctx.Err(); err != nil { + return Entry{}, err + } + normalized, err := healthDomain.NormalizeObservation(command.Observation) + if err != nil { + return Entry{}, err + } + if normalized.Level == healthDomain.LevelTarget || command.MaxConsecutiveFailures <= 0 { + return Entry{}, ErrInvalidHealthUpdate + } + if p == nil { + return Entry{}, ErrInvalidHealthUpdate + } + p.mu.Lock() + defer p.mu.Unlock() + if err := ctx.Err(); err != nil { + return Entry{}, err + } + p.purgeExpiredLocked(normalized.ObservedAt) + entry, ok := p.entryByIDLocked(normalized.ProxyID) + if !ok { + return Entry{}, ErrActivityNotFound + } + current := entry.GlobalHealth + if current.State == "" { + current.State = entry.State + } + if current.State != entry.State { + return Entry{}, ErrInvalidHealthUpdate + } + next, err := healthDomain.ReduceGlobal(current, normalized, command.MaxConsecutiveFailures) + if err != nil { + return Entry{}, err + } + if entry.State != next.State && !proxyDomain.CanTransition(entry.State, next.State) { + return Entry{}, ErrInvalidHealthUpdate + } + entry.State = next.State + entry.Proxy.State = next.State + entry.Proxy.LastCheckedAt = timePointer(normalized.ObservedAt) + entry.Proxy.Latency = normalized.Latency + if normalized.Success { + entry.Proxy.LastSuccessAt = timePointer(normalized.ObservedAt) + } + entry.GlobalHealth = next + p.setEntryByIDLocked(normalized.ProxyID, entry) + entry.Proxy = cloneProxy(entry.Proxy) + return entry, nil +} + func (p *MemoryPool) Inventory(ctx context.Context, upstreamID string, now time.Time) (Inventory, error) { result := Inventory{UpstreamID: upstreamID} if ctx == nil { @@ -477,6 +554,7 @@ func (p *MemoryPool) Extract(ctx context.Context, command extractionDomain.Comma entry := p.entries[key] entry.State = proxyDomain.StateExtracted entry.Proxy.State = proxyDomain.StateExtracted + entry.GlobalHealth.State = proxyDomain.StateExtracted p.entries[key] = entry result.Items = append(result.Items, extractionCandidate(entry)) } @@ -488,6 +566,11 @@ func (p *MemoryPool) Extract(ctx context.Context, command extractionDomain.Comma return result, nil } +func timePointer(value time.Time) *time.Time { + canonical := value.UTC() + return &canonical +} + func (p *MemoryPool) rememberExtractionLocked( idempotencyKey string, command extractionDomain.Command, diff --git a/internal/domain/health/health.go b/internal/domain/health/health.go new file mode 100644 index 0000000..ee6b5ab --- /dev/null +++ b/internal/domain/health/health.go @@ -0,0 +1,364 @@ +// Package health defines transport-free facts and deterministic health +// reduction rules shared by the Controller and Checker integration layers. +package health + +import ( + "crypto/sha256" + "errors" + "fmt" + "net/netip" + "net/url" + "strings" + "time" + + proxyDomain "proxy-pool/internal/domain/proxy" +) + +var ( + ErrInvalidObservation = errors.New("invalid health observation") + ErrInvalidGlobalState = errors.New("invalid global health state") + ErrInvalidTargetState = errors.New("invalid target health state") + ErrNonGlobalObservation = errors.New("target health observation cannot change global proxy health") + ErrNonTargetObservation = errors.New("global health observation cannot change target profile health") + ErrStaleObservation = errors.New("stale health observation") + ErrConflictingObservation = errors.New("conflicting health observation replay") + ErrInvalidFailureThreshold = errors.New("invalid health failure threshold") + ErrInvalidCheckPreparation = errors.New("invalid health check preparation") +) + +type Level string + +const ( + LevelBasic Level = "BASIC" + LevelEgress Level = "EGRESS" + LevelTarget Level = "TARGET" +) + +type TargetStatus string + +const ( + TargetUnknown TargetStatus = "UNKNOWN" + TargetAvailable TargetStatus = "AVAILABLE" + TargetSuspect TargetStatus = "SUSPECT" + TargetUnhealthy TargetStatus = "UNHEALTHY" +) + +// Observation is an immutable result produced by a Checker. It has no +// authority to modify a Proxy until the Controller reduces and commits it. +type Observation struct { + TaskID string + ProxyID string + Level Level + RoutingName string + TargetURL string + Success bool + FailureClass string + Latency time.Duration + ObservedEgressIP string + ObservedAt time.Time +} + +type TargetProfile struct { + RoutingName string + TargetURL string +} + +// GlobalState is the small authoritative state needed to reduce BASIC and +// EGRESS observations. The persistence adapter owns storing it atomically. +type GlobalState struct { + State proxyDomain.State + ConsecutiveFailures int + LastTaskID string + LastObservedAt time.Time + LastObservationDigest [sha256.Size]byte +} + +// TargetState is isolated by (proxy, routing, target URL). A target failure +// never changes a Proxy's GlobalState. +type TargetState struct { + Status TargetStatus + ConsecutiveFailures int + LastTaskID string + LastObservedAt time.Time + LastSuccessAt time.Time + Latency time.Duration + LastObservationDigest [sha256.Size]byte +} + +// NormalizeObservation validates and canonicalizes an observation before it +// crosses an idempotency or persistence boundary. +func NormalizeObservation(value Observation) (Observation, error) { + value.TaskID = strings.TrimSpace(value.TaskID) + value.ProxyID = strings.TrimSpace(value.ProxyID) + value.RoutingName = strings.TrimSpace(value.RoutingName) + value.TargetURL = strings.TrimSpace(value.TargetURL) + value.FailureClass = strings.TrimSpace(value.FailureClass) + value.ObservedEgressIP = strings.TrimSpace(value.ObservedEgressIP) + if !validIdentifier(value.TaskID) || !validIdentifier(value.ProxyID) || value.ObservedAt.IsZero() || value.Latency < 0 { + return Observation{}, ErrInvalidObservation + } + if value.Success && value.FailureClass != "" { + return Observation{}, ErrInvalidObservation + } + if value.ObservedEgressIP != "" { + if _, err := netip.ParseAddr(value.ObservedEgressIP); err != nil { + return Observation{}, ErrInvalidObservation + } + } + switch value.Level { + case LevelBasic, LevelEgress: + if value.RoutingName != "" || value.TargetURL != "" { + return Observation{}, ErrInvalidObservation + } + case LevelTarget: + profile, err := NormalizeTargetProfile(TargetProfile{RoutingName: value.RoutingName, TargetURL: value.TargetURL}) + if err != nil { + return Observation{}, err + } + value.RoutingName = profile.RoutingName + value.TargetURL = profile.TargetURL + default: + return Observation{}, ErrInvalidObservation + } + value.ObservedAt = value.ObservedAt.UTC() + return value, nil +} + +func NormalizeTargetProfile(value TargetProfile) (TargetProfile, error) { + value.RoutingName = strings.TrimSpace(value.RoutingName) + value.TargetURL = strings.TrimSpace(value.TargetURL) + if !validIdentifier(value.RoutingName) || value.TargetURL == "" { + return TargetProfile{}, ErrInvalidObservation + } + parsed, err := url.Parse(value.TargetURL) + if err != nil || parsed.Scheme == "" || parsed.Host == "" || parsed.User != nil || parsed.Fragment != "" { + return TargetProfile{}, ErrInvalidObservation + } + value.TargetURL = parsed.String() + return value, nil +} + +func (value TargetProfile) Key() string { + return value.RoutingName + "\x00" + value.TargetURL +} + +func ObservationDigest(value Observation) ([sha256.Size]byte, error) { + normalized, err := NormalizeObservation(value) + if err != nil { + return [sha256.Size]byte{}, err + } + payload := strings.Join([]string{ + normalized.TaskID, normalized.ProxyID, string(normalized.Level), normalized.RoutingName, + normalized.TargetURL, fmt.Sprintf("%t", normalized.Success), normalized.FailureClass, + normalized.Latency.String(), normalized.ObservedEgressIP, normalized.ObservedAt.Format(time.RFC3339Nano), + }, "\x00") + return sha256.Sum256([]byte(payload)), nil +} + +// BeginGlobalCheck moves only states that are deliberately unavailable during +// probing into CHECKING. AVAILABLE and SUSPECT remain serving states while a +// periodic probe is in flight. +func BeginGlobalCheck(current GlobalState) (proxyDomain.State, error) { + if err := validateGlobalState(current); err != nil { + return "", err + } + switch current.State { + case proxyDomain.StateFetched, proxyDomain.StateUnhealthy: + return proxyDomain.StateChecking, nil + case proxyDomain.StateChecking, proxyDomain.StateAvailable, proxyDomain.StateSuspect: + return current.State, nil + default: + return "", ErrInvalidCheckPreparation + } +} + +// ReduceGlobal reduces one BASIC or EGRESS fact. It is intentionally pure so +// Memory and Redis adapters can apply the same result atomically. +func ReduceGlobal(current GlobalState, observation Observation, maxConsecutiveFailures int) (GlobalState, error) { + if maxConsecutiveFailures <= 0 { + return GlobalState{}, ErrInvalidFailureThreshold + } + if err := validateGlobalState(current); err != nil { + return GlobalState{}, err + } + normalized, err := NormalizeObservation(observation) + if err != nil { + return GlobalState{}, err + } + if normalized.Level == LevelTarget { + return GlobalState{}, ErrNonGlobalObservation + } + digest, err := ObservationDigest(normalized) + if err != nil { + return GlobalState{}, err + } + if err := checkObservationOrder(current.LastTaskID, current.LastObservedAt, current.LastObservationDigest, normalized, digest); err != nil { + if errors.Is(err, errExactReplay) { + return current, nil + } + return GlobalState{}, err + } + next := current + next.LastTaskID = normalized.TaskID + next.LastObservedAt = normalized.ObservedAt + next.LastObservationDigest = digest + if normalized.Success { + switch current.State { + case proxyDomain.StateChecking, proxyDomain.StateAvailable, proxyDomain.StateSuspect: + next.State = proxyDomain.StateAvailable + next.ConsecutiveFailures = 0 + return next, nil + default: + return GlobalState{}, ErrInvalidGlobalState + } + } + switch current.State { + case proxyDomain.StateChecking: + next.ConsecutiveFailures++ + next.State = proxyDomain.StateUnhealthy + return next, nil + case proxyDomain.StateAvailable, proxyDomain.StateSuspect: + next.ConsecutiveFailures++ + if next.ConsecutiveFailures >= maxConsecutiveFailures { + next.State = proxyDomain.StateUnhealthy + } else { + next.State = proxyDomain.StateSuspect + } + return next, nil + default: + return GlobalState{}, ErrInvalidGlobalState + } +} + +// ReduceTarget reduces one TARGET fact into only its target profile state. +// Callers must keep this result separate from activity-pool global state. +func ReduceTarget(current TargetState, observation Observation, maxConsecutiveFailures int) (TargetState, error) { + if maxConsecutiveFailures <= 0 { + return TargetState{}, ErrInvalidFailureThreshold + } + if current.Status == "" { + current.Status = TargetUnknown + } + if err := validateTargetState(current); err != nil { + return TargetState{}, err + } + normalized, err := NormalizeObservation(observation) + if err != nil { + return TargetState{}, err + } + if normalized.Level != LevelTarget { + return TargetState{}, ErrNonTargetObservation + } + digest, err := ObservationDigest(normalized) + if err != nil { + return TargetState{}, err + } + if err := checkObservationOrder(current.LastTaskID, current.LastObservedAt, current.LastObservationDigest, normalized, digest); err != nil { + if errors.Is(err, errExactReplay) { + return current, nil + } + return TargetState{}, err + } + next := current + next.LastTaskID = normalized.TaskID + next.LastObservedAt = normalized.ObservedAt + next.LastObservationDigest = digest + next.Latency = normalized.Latency + if normalized.Success { + next.Status = TargetAvailable + next.ConsecutiveFailures = 0 + next.LastSuccessAt = normalized.ObservedAt + return next, nil + } + next.ConsecutiveFailures++ + if next.ConsecutiveFailures >= maxConsecutiveFailures { + next.Status = TargetUnhealthy + } else { + next.Status = TargetSuspect + } + return next, nil +} + +var errExactReplay = errors.New("exact health observation replay") + +func checkObservationOrder( + lastTaskID string, + lastObservedAt time.Time, + lastDigest [sha256.Size]byte, + observation Observation, + digest [sha256.Size]byte, +) error { + if lastObservedAt.IsZero() { + return nil + } + if observation.ObservedAt.Before(lastObservedAt) { + return ErrStaleObservation + } + if observation.TaskID == lastTaskID { + if digest == lastDigest { + return errExactReplay + } + return ErrConflictingObservation + } + if observation.ObservedAt.Equal(lastObservedAt) { + return ErrStaleObservation + } + return nil +} + +func validateGlobalState(value GlobalState) error { + if value.ConsecutiveFailures < 0 { + return ErrInvalidGlobalState + } + switch value.State { + case proxyDomain.StateFetched, proxyDomain.StateChecking, proxyDomain.StateAvailable, + proxyDomain.StateSuspect, proxyDomain.StateUnhealthy: + default: + return ErrInvalidGlobalState + } + if value.LastObservedAt.IsZero() { + if value.LastTaskID != "" || value.ConsecutiveFailures != 0 || value.LastObservationDigest != ([sha256.Size]byte{}) { + return ErrInvalidGlobalState + } + return nil + } + if !validIdentifier(value.LastTaskID) || value.LastObservationDigest == ([sha256.Size]byte{}) { + return ErrInvalidGlobalState + } + return nil +} + +func validateTargetState(value TargetState) error { + if value.ConsecutiveFailures < 0 || value.Latency < 0 { + return ErrInvalidTargetState + } + switch value.Status { + case TargetUnknown, TargetAvailable, TargetSuspect, TargetUnhealthy: + default: + return ErrInvalidTargetState + } + if value.LastObservedAt.IsZero() { + if value.Status != TargetUnknown || value.LastTaskID != "" || value.ConsecutiveFailures != 0 || + !value.LastSuccessAt.IsZero() || value.LastObservationDigest != ([sha256.Size]byte{}) { + return ErrInvalidTargetState + } + return nil + } + if !validIdentifier(value.LastTaskID) || value.LastObservationDigest == ([sha256.Size]byte{}) || + (!value.LastSuccessAt.IsZero() && value.LastSuccessAt.After(value.LastObservedAt)) { + return ErrInvalidTargetState + } + return nil +} + +func validIdentifier(value string) bool { + if value == "" || len(value) > 256 { + return false + } + for _, character := range value { + if character <= ' ' || character == '\x7f' { + return false + } + } + return true +} diff --git a/internal/domain/health/health_test.go b/internal/domain/health/health_test.go new file mode 100644 index 0000000..bc3f8a1 --- /dev/null +++ b/internal/domain/health/health_test.go @@ -0,0 +1,95 @@ +package health + +import ( + "errors" + "testing" + "time" + + proxyDomain "proxy-pool/internal/domain/proxy" +) + +func TestReduceGlobalAppliesSuspectUnhealthyAndRecovery(t *testing.T) { + now := time.Date(2026, 7, 31, 10, 0, 0, 0, time.UTC) + current := GlobalState{State: proxyDomain.StateAvailable} + first, err := ReduceGlobal(current, globalObservation("task-1", false, now), 2) + if err != nil || first.State != proxyDomain.StateSuspect || first.ConsecutiveFailures != 1 { + t.Fatalf("first failure = %+v, %v; want SUSPECT / 1", first, err) + } + second, err := ReduceGlobal(first, globalObservation("task-2", false, now.Add(time.Second)), 2) + if err != nil || second.State != proxyDomain.StateUnhealthy || second.ConsecutiveFailures != 2 { + t.Fatalf("second failure = %+v, %v; want UNHEALTHY / 2", second, err) + } + if next, err := BeginGlobalCheck(second); err != nil || next != proxyDomain.StateChecking { + t.Fatalf("BeginGlobalCheck(unhealthy) = %s, %v; want CHECKING", next, err) + } + recovering := second + recovering.State = proxyDomain.StateChecking + recovered, err := ReduceGlobal(recovering, globalObservation("task-3", true, now.Add(2*time.Second)), 2) + if err != nil || recovered.State != proxyDomain.StateAvailable || recovered.ConsecutiveFailures != 0 { + t.Fatalf("recovery = %+v, %v; want AVAILABLE / 0", recovered, err) + } +} + +func TestReduceGlobalMarksFailedInitialCheckUnhealthy(t *testing.T) { + now := time.Date(2026, 7, 31, 10, 0, 0, 0, time.UTC) + next, err := ReduceGlobal(GlobalState{State: proxyDomain.StateChecking}, globalObservation("task-1", false, now), 3) + if err != nil || next.State != proxyDomain.StateUnhealthy || next.ConsecutiveFailures != 1 { + t.Fatalf("initial failure = %+v, %v; want UNHEALTHY / 1", next, err) + } +} + +func TestReduceGlobalRejectsStaleAndConflictingReplays(t *testing.T) { + now := time.Date(2026, 7, 31, 10, 0, 0, 0, time.UTC) + first, err := ReduceGlobal(GlobalState{State: proxyDomain.StateAvailable}, globalObservation("task-1", false, now), 2) + if err != nil { + t.Fatalf("ReduceGlobal(first): %v", err) + } + replayed, err := ReduceGlobal(first, globalObservation("task-1", false, now), 2) + if err != nil || replayed != first { + t.Fatalf("ReduceGlobal(replay) = %+v, %v; want unchanged", replayed, err) + } + if _, err := ReduceGlobal(first, globalObservation("task-1", true, now), 2); !errors.Is(err, ErrConflictingObservation) { + t.Fatalf("conflicting replay error = %v, want ErrConflictingObservation", err) + } + if _, err := ReduceGlobal(first, globalObservation("task-2", true, now.Add(-time.Second)), 2); !errors.Is(err, ErrStaleObservation) { + t.Fatalf("stale observation error = %v, want ErrStaleObservation", err) + } +} + +func TestReduceTargetNeverChangesGlobalState(t *testing.T) { + now := time.Date(2026, 7, 31, 10, 0, 0, 0, time.UTC) + observation := Observation{ + TaskID: "task-target-1", ProxyID: "proxy-a", Level: LevelTarget, RoutingName: "route-a", + TargetURL: "https://target.example/path?volatile=true", Success: false, FailureClass: "target_403", + Latency: 20 * time.Millisecond, ObservedAt: now, + } + target, err := ReduceTarget(TargetState{}, observation, 2) + if err != nil || target.Status != TargetSuspect || target.ConsecutiveFailures != 1 { + t.Fatalf("ReduceTarget(first) = %+v, %v; want SUSPECT / 1", target, err) + } + if _, err := ReduceGlobal(GlobalState{State: proxyDomain.StateAvailable}, observation, 2); !errors.Is(err, ErrNonGlobalObservation) { + t.Fatalf("ReduceGlobal(target) error = %v, want ErrNonGlobalObservation", err) + } + profile, err := NormalizeTargetProfile(TargetProfile{RoutingName: observation.RoutingName, TargetURL: observation.TargetURL}) + if err != nil || profile.TargetURL != "https://target.example/path?volatile=true" { + t.Fatalf("NormalizeTargetProfile() = %+v, %v", profile, err) + } +} + +func TestNormalizeObservationRejectsCrossLevelFields(t *testing.T) { + now := time.Date(2026, 7, 31, 10, 0, 0, 0, time.UTC) + _, err := NormalizeObservation(Observation{ + TaskID: "task-a", ProxyID: "proxy-a", Level: LevelBasic, RoutingName: "route-a", Success: true, ObservedAt: now, + }) + if !errors.Is(err, ErrInvalidObservation) { + t.Fatalf("NormalizeObservation() error = %v, want ErrInvalidObservation", err) + } +} + +func globalObservation(taskID string, success bool, observedAt time.Time) Observation { + observation := Observation{TaskID: taskID, ProxyID: "proxy-a", Level: LevelBasic, Success: success, ObservedAt: observedAt} + if !success { + observation.FailureClass = "timeout" + } + return observation +}