From 19cf3e32fa595ffbee7b31f77b269b953471c64f Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 31 Jul 2026 18:04:05 +0800 Subject: [PATCH] feat: persist target health profiles --- docs/development/implementation-plan.md | 11 +- docs/requirements/completion-audit.md | 7 +- docs/requirements/traceability.md | 2 +- .../adapters/redisactivity/adapter_test.go | 1 + internal/adapters/redisactivity/health.go | 134 ++++++++++++++++++ internal/adapters/redisactivity/keys.go | 7 + internal/adapters/redisactivity/scripts.go | 9 ++ .../redisactivity/scripts/target_health.lua | 116 +++++++++++++++ .../target_health_integration_test.go | 91 ++++++++++++ internal/domain/activitypool/pool.go | 68 +++++++++ .../domain/activitypool/target_health_test.go | 43 ++++++ 11 files changed, 480 insertions(+), 9 deletions(-) create mode 100644 internal/adapters/redisactivity/scripts/target_health.lua create mode 100644 internal/adapters/redisactivity/target_health_integration_test.go create mode 100644 internal/domain/activitypool/target_health_test.go diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 2f78594..d9b9a74 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -273,11 +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 或独立进程，因此本任务保持未完成。 +当前进度（2026-07-31）：已新增不依赖 gRPC 的 `domain/health` Observation、全局/Target +Profile Reducer 与任务摘要幂等语义；BASIC/EGRESS 结果经活动池窄接口在 Memory 和 Redis +Lua 同一原子边界归并，覆盖首次 CHECKING 失败进入 UNHEALTHY、AVAILABLE/SUSPECT 的 +阈值降级、精确重放、冲突拒绝与成功恢复。TARGET Profile 也已在 Memory 和 Redis 中独立 +归并，以哈希键保存并随代理 TTL 过期，绝不写入 Proxy 全局状态或选择索引。调度器、 +Checker RPC 与独立进程尚未接入，因此本任务保持未完成。 ## Task 12: Machine-readable Contracts diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 6bde411..5a6bfba 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -58,8 +58,8 @@ Routing 一起纳入 Snapshot checksum，仅在 mTLS 和 Gateway 当前内存 Vi Routing payload 已由配置顺序和 Admin 当前状态合成并覆盖 Snapshot checksum，其运行时消费、 Outcome 已实现为 Gateway 本地有界队列、微批确认重试和 Controller 的 session/sequence/ 摘要 Redis 栅栏；原始事件不落 Redis 或 PostgreSQL。Checker 已有全局健康 Reducer -与 Memory/Redis 原子状态提交基础，但任务调度、Checker RPC、TARGET Profile 持久化和 -独立执行进程尚未闭环。Snapshot 签发在 Redis 中原子匹配当前 +与 Memory/Redis 原子状态提交基础；TARGET Profile 以独立、随代理 TTL 过期的 Redis +记录归并，不改写 Proxy 全局状态。任务调度、Checker RPC 和独立执行进程尚未闭环。Snapshot 签发在 Redis 中原子匹配当前 `session_id`，重注册会清除旧引用，迟到旧 Stream 不会覆盖新 session。Controller 在最近成功下发的 Snapshot `valid_until` 到达时关闭流；Gateway 的公用 `SessionSupervisor` 已实现可恢复错误的有界退避重连。Gateway 会校验并执行 Snapshot @@ -109,7 +109,8 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 6. Worker 基础网络快照流、Proxy/Gateway Routing/凭据 Snapshot payload、Gateway Snapshot 客户端和进程装配、同版本 Routing 编译/动态匹配、五种策略上游选择与 reject/wait/direct 已完成； Outcome 上报已完成基础观测链；Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。 -7. Checker 调度、探测器和健康 reducer。 +7. Checker 调度、探测器和健康 reducer；其中全局与 TARGET Profile 的 Memory/Redis + 原子归并已完成。 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 10. 将 reject/wait/direct 接入 Distribution 运行链，补齐 Sequential 持久化恢复、跨实例 CAS diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index 88fa4b3..ebb4671 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -79,7 +79,7 @@ | ID | 最终需求 | 来源 | 验证证据 | |---|---|---|---| -| HEALTH-001 | 全局健康与 Routing/目标健康分离 | 221-270, 8679-8708 | `domain/health` 已将 BASIC/EGRESS 全局 Reducer 与 TARGET Profile Reducer 分离；TARGET 持久化和 Routing 消费待实现 | +| 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 编排待实现 | | SEC-001 | API 认证与 Proxy 认证分离，Secret 统一脱敏 | 7528-8111, 8904-8945 | Config 脱敏、Provider Store -> SecretRef -> Gateway Resolver 跨包测试与格式化泄漏回归测试 | diff --git a/internal/adapters/redisactivity/adapter_test.go b/internal/adapters/redisactivity/adapter_test.go index 67e0088..bb0c96f 100644 --- a/internal/adapters/redisactivity/adapter_test.go +++ b/internal/adapters/redisactivity/adapter_test.go @@ -110,6 +110,7 @@ func TestNewBuildsClusterSafeKeyspaceAndHashesDynamicTokens(t *testing.T) { dynamicKeys := []string{ adapter.keys.idempotency(raw, raw), adapter.keys.operation(raw), + adapter.keys.targetHealth(raw), adapter.keys.protocol(raw), adapter.keys.region(raw), adapter.keys.carrier(raw), diff --git a/internal/adapters/redisactivity/health.go b/internal/adapters/redisactivity/health.go index f04dade..15f8e26 100644 --- a/internal/adapters/redisactivity/health.go +++ b/internal/adapters/redisactivity/health.go @@ -2,7 +2,10 @@ package redisactivity import ( "context" + "crypto/sha256" "encoding/hex" + "errors" + "time" "proxy-pool/internal/domain/activitypool" healthDomain "proxy-pool/internal/domain/health" @@ -10,6 +13,18 @@ import ( var _ activitypool.HealthStore = (*Adapter)(nil) var _ activitypool.GlobalHealthStore = (*Adapter)(nil) +var _ activitypool.TargetHealthStore = (*Adapter)(nil) + +type targetHealthRecord struct { + Version int `json:"version"` + Status string `json:"status"` + ConsecutiveFailures int `json:"consecutiveFailures"` + LastHealthTaskID string `json:"lastHealthTaskId,omitempty"` + LastHealthDigest string `json:"lastHealthDigest,omitempty"` + LastHealthObservedAtMS int64 `json:"lastHealthObservedAtMs,omitempty"` + LastSuccessAtMS int64 `json:"lastSuccessAtMs,omitempty"` + LatencyNS int64 `json:"latencyNs"` +} func (a *Adapter) ApplyHealth(ctx context.Context, update activitypool.HealthUpdate) (activitypool.Entry, error) { if ctx == nil { @@ -127,3 +142,122 @@ func (a *Adapter) ApplyGlobalObservation( return activitypool.Entry{}, invalidScriptReply("unexpected global health status") } } + +// ApplyTargetObservation reduces a TARGET check into a short-lived Profile +// record keyed by (proxy, routing, target URL). It intentionally does not +// update the proxy record, global health, or any proxy selection index. +func (a *Adapter) ApplyTargetObservation( + ctx context.Context, + command activitypool.TargetHealthCommand, +) (healthDomain.TargetState, error) { + if ctx == nil { + return healthDomain.TargetState{}, activitypool.ErrInvalidHealthUpdate + } + if err := ctx.Err(); err != nil { + return healthDomain.TargetState{}, err + } + normalized, err := healthDomain.NormalizeObservation(command.Observation) + if err != nil { + return healthDomain.TargetState{}, err + } + if a == nil || normalized.Level != healthDomain.LevelTarget || command.MaxConsecutiveFailures <= 0 { + return healthDomain.TargetState{}, activitypool.ErrInvalidHealthUpdate + } + profile, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{ + RoutingName: normalized.RoutingName, + TargetURL: normalized.TargetURL, + }) + if err != nil { + return healthDomain.TargetState{}, err + } + digest, err := healthDomain.ObservationDigest(normalized) + if err != nil { + return healthDomain.TargetState{}, err + } + operationID, err := newOperationID() + if err != nil { + return healthDomain.TargetState{}, err + } + success := "0" + if normalized.Success { + success = "1" + } + result, err := runScript(ctx, a.client, targetHealthScript, []string{ + a.keys.records, a.keys.targetHealth(normalized.ProxyID), a.keys.operation(operationID), + }, normalized.ObservedAt.UnixMilli(), int64(normalized.Latency), normalized.ProxyID, + digestToken(profile.Key()), success, command.MaxConsecutiveFailures, normalized.TaskID, + hex.EncodeToString(digest[:]), operationTTLMillis(a.options.OperationTTL)) + if err != nil { + return healthDomain.TargetState{}, err + } + var reply targetHealthScriptReply + if err := decodeScriptResult(result, &reply); err != nil { + return healthDomain.TargetState{}, err + } + switch reply.Status { + case scriptNotFound: + return healthDomain.TargetState{}, activitypool.ErrActivityNotFound + case scriptStale: + return healthDomain.TargetState{}, healthDomain.ErrStaleObservation + case scriptConflict: + return healthDomain.TargetState{}, healthDomain.ErrConflictingObservation + case scriptInvalid: + return healthDomain.TargetState{}, activitypool.ErrInvalidHealthUpdate + case scriptOK: + if reply.Target == "" { + return healthDomain.TargetState{}, invalidScriptReply("target health reply omitted target state") + } + state, err := decodeTargetHealthState(reply.Target) + if err != nil { + return healthDomain.TargetState{}, invalidScriptReply("target health reply contained an invalid target state") + } + return state, nil + default: + return healthDomain.TargetState{}, invalidScriptReply("unexpected target health status") + } +} + +func decodeTargetHealthState(payload string) (healthDomain.TargetState, error) { + var record targetHealthRecord + if err := decodeJSON(payload, &record); err != nil { + return healthDomain.TargetState{}, err + } + if record.Version != recordVersion || record.ConsecutiveFailures < 0 || record.LatencyNS < 0 { + return healthDomain.TargetState{}, ErrInvalidRecord + } + state := healthDomain.TargetState{ + Status: healthDomain.TargetStatus(record.Status), + ConsecutiveFailures: record.ConsecutiveFailures, + LastTaskID: record.LastHealthTaskID, + Latency: time.Duration(record.LatencyNS), + } + switch state.Status { + case healthDomain.TargetUnknown, healthDomain.TargetAvailable, healthDomain.TargetSuspect, healthDomain.TargetUnhealthy: + default: + return healthDomain.TargetState{}, ErrInvalidRecord + } + if record.LastHealthObservedAtMS == 0 { + if state.Status != healthDomain.TargetUnknown || state.LastTaskID != "" || state.ConsecutiveFailures != 0 || + record.LastHealthDigest != "" || record.LastSuccessAtMS != 0 || state.Latency != 0 { + return healthDomain.TargetState{}, ErrInvalidRecord + } + return state, nil + } + if record.LastHealthObservedAtMS < 0 || record.LastSuccessAtMS < 0 || state.LastTaskID == "" || + len(record.LastHealthDigest) != sha256.Size*2 { + return healthDomain.TargetState{}, ErrInvalidRecord + } + digest, err := hex.DecodeString(record.LastHealthDigest) + if err != nil || len(digest) != sha256.Size { + return healthDomain.TargetState{}, errors.Join(ErrInvalidRecord, err) + } + copy(state.LastObservationDigest[:], digest) + state.LastObservedAt = time.UnixMilli(record.LastHealthObservedAtMS).UTC() + if record.LastSuccessAtMS > 0 { + state.LastSuccessAt = time.UnixMilli(record.LastSuccessAtMS).UTC() + if state.LastSuccessAt.After(state.LastObservedAt) { + return healthDomain.TargetState{}, ErrInvalidRecord + } + } + return state, nil +} diff --git a/internal/adapters/redisactivity/keys.go b/internal/adapters/redisactivity/keys.go index ab602b5..5702f0d 100644 --- a/internal/adapters/redisactivity/keys.go +++ b/internal/adapters/redisactivity/keys.go @@ -65,6 +65,13 @@ func (keys keyspace) operation(operationID string) string { return keys.prefix + ":op:" + digestToken(operationID) } +// targetHealth stores short-lived TARGET check state separately from the +// proxy record. The proxy ID is hashed so neither endpoint identity nor target +// profile information appears in a Redis key name. +func (keys keyspace) targetHealth(proxyID string) string { + return keys.prefix + ":target-health:" + digestToken(proxyID) +} + func (keys keyspace) protocol(value string) string { return keys.facet("protocol", value) } diff --git a/internal/adapters/redisactivity/scripts.go b/internal/adapters/redisactivity/scripts.go index 6453a94..79ebf1f 100644 --- a/internal/adapters/redisactivity/scripts.go +++ b/internal/adapters/redisactivity/scripts.go @@ -44,6 +44,11 @@ type healthScriptReply struct { Record string `json:"record,omitempty"` } +type targetHealthScriptReply struct { + Status scriptStatus `json:"status"` + Target string `json:"target,omitempty"` +} + type extractScriptReply struct { Status scriptStatus `json:"status"` RequestDigest string `json:"requestDigest"` @@ -105,6 +110,9 @@ var upsertSource string //go:embed scripts/health.lua var healthSource string +//go:embed scripts/target_health.lua +var targetHealthSource string + //go:embed scripts/extract.lua var extractSource string @@ -129,6 +137,7 @@ var workerSnapshotSource string var ( upsertScript = redis.NewScript(upsertSource) healthScript = redis.NewScript(healthSource) + targetHealthScript = redis.NewScript(targetHealthSource) extractScript = redis.NewScript(extractSource) ownershipScript = redis.NewScript(ownershipSource) sweepScript = redis.NewScript(sweepSource) diff --git a/internal/adapters/redisactivity/scripts/target_health.lua b/internal/adapters/redisactivity/scripts/target_health.lua new file mode 100644 index 0000000..a8a30c4 --- /dev/null +++ b/internal/adapters/redisactivity/scripts/target_health.lua @@ -0,0 +1,116 @@ +local records_key = KEYS[1] +local target_health_key = KEYS[2] +local operation_key = KEYS[3] + +local observed_at_ms = tonumber(ARGV[1]) +local latency_ns = tonumber(ARGV[2]) +local proxy_id = ARGV[3] +local profile_key = ARGV[4] +local success = ARGV[5] +local max_consecutive_failures = tonumber(ARGV[6]) +local task_id = ARGV[7] +local observation_digest = ARGV[8] +local operation_ttl_ms = tonumber(ARGV[9]) + +local committed = redis.call('GET', operation_key) +if committed then + return committed +end + +local function finish(reply, hard_expiry_ms) + local encoded = cjson.encode(reply) + redis.call('SET', operation_key, encoded, 'PX', operation_ttl_ms) + if hard_expiry_ms then + local operation_expiry_ms = redis.call('PEXPIRETIME', operation_key) + if operation_expiry_ms > hard_expiry_ms then + redis.call('PEXPIREAT', operation_key, hard_expiry_ms) + end + end + return encoded +end + +if not observed_at_ms or observed_at_ms <= 0 or not latency_ns or latency_ns < 0 or + type(proxy_id) ~= 'string' or proxy_id == '' or type(profile_key) ~= 'string' or + string.len(profile_key) ~= 64 or (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 or + not operation_ttl_ms or operation_ttl_ms <= 0 then + return finish({status = 'invalid'}) +end + +local raw_proxy = redis.call('HGET', records_key, proxy_id) +if not raw_proxy then + return finish({status = 'not_found'}) +end +local proxy = cjson.decode(raw_proxy) +if type(proxy.expiresAtMs) ~= 'number' or proxy.expiresAtMs <= observed_at_ms then + redis.call('DEL', target_health_key) + return finish({status = 'not_found'}) +end + +local raw_target = redis.call('HGET', target_health_key, profile_key) +local target +if raw_target then + local decoded, value = pcall(cjson.decode, raw_target) + if not decoded or type(value) ~= 'table' then + return finish({status = 'invalid'}) + end + target = value +else + target = { + version = 1, + status = 'UNKNOWN', + consecutiveFailures = 0, + latencyNs = 0, + } +end + +if target.version ~= 1 or type(target.status) ~= 'string' or + type(target.consecutiveFailures) ~= 'number' or target.consecutiveFailures < 0 or + type(target.latencyNs) ~= 'number' or target.latencyNs < 0 then + return finish({status = 'invalid'}) +end + +local last_task_id = target.lastHealthTaskId or '' +local last_digest = target.lastHealthDigest or '' +local last_observed_at_ms = tonumber(target.lastHealthObservedAtMs or '0') +if not last_observed_at_ms or last_observed_at_ms < 0 then + return finish({status = 'invalid'}) +end +if last_observed_at_ms > 0 then + if type(last_task_id) ~= 'string' or last_task_id == '' or type(last_digest) ~= 'string' or + string.len(last_digest) ~= 64 then + return finish({status = 'invalid'}) + end + if last_task_id == task_id then + if last_digest == observation_digest then + return finish({status = 'ok', target = raw_target}, proxy.expiresAtMs) + end + return finish({status = 'conflict'}) + end + if observed_at_ms <= last_observed_at_ms then + return finish({status = 'stale'}) + end +end + +target.lastHealthTaskId = task_id +target.lastHealthDigest = observation_digest +target.lastHealthObservedAtMs = observed_at_ms +target.latencyNs = latency_ns +if success == '1' then + target.status = 'AVAILABLE' + target.consecutiveFailures = 0 + target.lastSuccessAtMs = observed_at_ms +else + target.consecutiveFailures = target.consecutiveFailures + 1 + if target.consecutiveFailures >= max_consecutive_failures then + target.status = 'UNHEALTHY' + else + target.status = 'SUSPECT' + end +end + +local encoded = cjson.encode(target) +redis.call('HSET', target_health_key, profile_key, encoded) +redis.call('PEXPIREAT', target_health_key, proxy.expiresAtMs) +return finish({status = 'ok', target = encoded}, proxy.expiresAtMs) diff --git a/internal/adapters/redisactivity/target_health_integration_test.go b/internal/adapters/redisactivity/target_health_integration_test.go new file mode 100644 index 0000000..76c98bf --- /dev/null +++ b/internal/adapters/redisactivity/target_health_integration_test.go @@ -0,0 +1,91 @@ +//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 TestRedisKeepsTargetHealthOutsideGlobalProxyState(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) + } + firstObservation := redisTargetObservation("target-task-1", false, now.Add(time.Second)) + first, err := fixture.Adapter.ApplyTargetObservation(context.Background(), activitypool.TargetHealthCommand{ + Observation: firstObservation, MaxConsecutiveFailures: 2, + }) + if err != nil || first.Status != healthDomain.TargetSuspect || first.ConsecutiveFailures != 1 || + first.LastObservedAt != firstObservation.ObservedAt || first.Latency != firstObservation.Latency { + t.Fatalf("ApplyTargetObservation(first) = %+v, %v", first, err) + } + replayed, err := fixture.Adapter.ApplyTargetObservation(context.Background(), activitypool.TargetHealthCommand{ + Observation: firstObservation, MaxConsecutiveFailures: 2, + }) + if err != nil || replayed != first { + t.Fatalf("ApplyTargetObservation(replay) = %+v, %v", replayed, err) + } + conflicting := firstObservation + conflicting.Success = true + conflicting.FailureClass = "" + if _, err := fixture.Adapter.ApplyTargetObservation(context.Background(), activitypool.TargetHealthCommand{ + Observation: conflicting, MaxConsecutiveFailures: 2, + }); !errors.Is(err, healthDomain.ErrConflictingObservation) { + t.Fatalf("ApplyTargetObservation(conflict) error = %v, want ErrConflictingObservation", err) + } + second, err := fixture.Adapter.ApplyTargetObservation(context.Background(), activitypool.TargetHealthCommand{ + Observation: redisTargetObservation("target-task-2", false, now.Add(2*time.Second)), MaxConsecutiveFailures: 2, + }) + if err != nil || second.Status != healthDomain.TargetUnhealthy || second.ConsecutiveFailures != 2 { + t.Fatalf("ApplyTargetObservation(second failure) = %+v, %v", second, err) + } + recoveredObservation := redisTargetObservation("target-task-3", true, now.Add(3*time.Second)) + recovered, err := fixture.Adapter.ApplyTargetObservation(context.Background(), activitypool.TargetHealthCommand{ + Observation: recoveredObservation, MaxConsecutiveFailures: 2, + }) + if err != nil || recovered.Status != healthDomain.TargetAvailable || recovered.ConsecutiveFailures != 0 || + recovered.LastSuccessAt != recoveredObservation.ObservedAt { + t.Fatalf("ApplyTargetObservation(recovery) = %+v, %v", recovered, err) + } + + profile := healthDomain.TargetProfile{RoutingName: recoveredObservation.RoutingName, TargetURL: recoveredObservation.TargetURL} + profileKey := fixture.Adapter.keys.targetHealth("proxy-a") + field := digestToken(profile.Key()) + if ttl, err := fixture.Client.PTTL(context.Background(), profileKey).Result(); err != nil || ttl <= 0 { + t.Fatalf("target health key TTL = (%v, %v), want positive", ttl, err) + } + if exists, err := fixture.Client.HExists(context.Background(), profileKey, field).Result(); err != nil || !exists { + t.Fatalf("target health profile exists = (%v, %v), want true", exists, err) + } + + global, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{ + ProxyID: "proxy-a", CheckedAt: now.Add(4 * time.Second), NextState: proxyDomain.StateChecking, + }) + if err != nil || global.State != proxyDomain.StateChecking || global.GlobalHealth.LastObservedAt != (time.Time{}) || + global.GlobalHealth.ConsecutiveFailures != 0 { + t.Fatalf("target observation changed global proxy state: %+v, %v", global, err) + } +} + +func redisTargetObservation(taskID string, success bool, observedAt time.Time) healthDomain.Observation { + observation := healthDomain.Observation{ + TaskID: taskID, ProxyID: "proxy-a", Level: healthDomain.LevelTarget, + RoutingName: "route-a", TargetURL: "https://target.example/check", Success: success, + Latency: 20 * time.Millisecond, ObservedAt: observedAt, + } + if !success { + observation.FailureClass = "target_403" + } + return observation +} diff --git a/internal/domain/activitypool/pool.go b/internal/domain/activitypool/pool.go index 9118071..8206a34 100644 --- a/internal/domain/activitypool/pool.go +++ b/internal/domain/activitypool/pool.go @@ -90,6 +90,17 @@ type GlobalHealthStore interface { ApplyGlobalObservation(context.Context, GlobalHealthCommand) (Entry, error) } +// TargetHealthCommand carries a TARGET fact. Its result is isolated by Proxy +// and TargetProfile and must not modify the activity-pool Proxy state. +type TargetHealthCommand struct { + Observation healthDomain.Observation + MaxConsecutiveFailures int +} + +type TargetHealthStore interface { + ApplyTargetObservation(context.Context, TargetHealthCommand) (healthDomain.TargetState, error) +} + type InventoryReader interface { Inventory(context.Context, string, time.Time) (Inventory, error) } @@ -117,9 +128,15 @@ type MemoryPool struct { keyByID map[string]string idempotent map[string]idempotencyEntry ownership map[string]ownershipDomain.Assignment + targets map[targetHealthKey]healthDomain.TargetState nextEpoch uint64 } +type targetHealthKey struct { + proxyID string + profileKey string +} + type idempotencyEntry struct { command extractionDomain.Command result extractionDomain.Result @@ -130,6 +147,7 @@ var ( _ Upserter = (*MemoryPool)(nil) _ HealthStore = (*MemoryPool)(nil) _ GlobalHealthStore = (*MemoryPool)(nil) + _ TargetHealthStore = (*MemoryPool)(nil) _ InventoryReader = (*MemoryPool)(nil) _ StateInventoryReader = (*MemoryPool)(nil) _ Maintainer = (*MemoryPool)(nil) @@ -143,6 +161,7 @@ func NewMemoryPool() *MemoryPool { keyByID: make(map[string]string), idempotent: make(map[string]idempotencyEntry), ownership: make(map[string]ownershipDomain.Assignment), + targets: make(map[targetHealthKey]healthDomain.TargetState), } } @@ -372,6 +391,50 @@ func (p *MemoryPool) ApplyGlobalObservation( return entry, nil } +// ApplyTargetObservation atomically reduces an isolated TARGET profile. It +// intentionally leaves the global Proxy state, health timestamp and indexes +// untouched. +func (p *MemoryPool) ApplyTargetObservation( + ctx context.Context, + command TargetHealthCommand, +) (healthDomain.TargetState, error) { + if ctx == nil { + return healthDomain.TargetState{}, ErrInvalidHealthUpdate + } + if err := ctx.Err(); err != nil { + return healthDomain.TargetState{}, err + } + normalized, err := healthDomain.NormalizeObservation(command.Observation) + if err != nil { + return healthDomain.TargetState{}, err + } + if normalized.Level != healthDomain.LevelTarget || command.MaxConsecutiveFailures <= 0 || p == nil { + return healthDomain.TargetState{}, ErrInvalidHealthUpdate + } + profile, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{ + RoutingName: normalized.RoutingName, TargetURL: normalized.TargetURL, + }) + if err != nil { + return healthDomain.TargetState{}, err + } + p.mu.Lock() + defer p.mu.Unlock() + if err := ctx.Err(); err != nil { + return healthDomain.TargetState{}, err + } + p.purgeExpiredLocked(normalized.ObservedAt) + if _, exists := p.entryByIDLocked(normalized.ProxyID); !exists { + return healthDomain.TargetState{}, ErrActivityNotFound + } + key := targetHealthKey{proxyID: normalized.ProxyID, profileKey: profile.Key()} + next, err := healthDomain.ReduceTarget(p.targets[key], normalized, command.MaxConsecutiveFailures) + if err != nil { + return healthDomain.TargetState{}, err + } + p.targets[key] = next + return next, nil +} + func (p *MemoryPool) Inventory(ctx context.Context, upstreamID string, now time.Time) (Inventory, error) { result := Inventory{UpstreamID: upstreamID} if ctx == nil { @@ -807,6 +870,11 @@ func (p *MemoryPool) removeEntryByIDLocked(proxyID string) { func (p *MemoryPool) removeEntryLocked(key string, entry Entry) { delete(p.ownership, entry.Proxy.ID) + for target := range p.targets { + if target.proxyID == entry.Proxy.ID { + delete(p.targets, target) + } + } delete(p.keyByID, entry.Proxy.ID) delete(p.entries, key) } diff --git a/internal/domain/activitypool/target_health_test.go b/internal/domain/activitypool/target_health_test.go new file mode 100644 index 0000000..70a1273 --- /dev/null +++ b/internal/domain/activitypool/target_health_test.go @@ -0,0 +1,43 @@ +package activitypool + +import ( + "context" + "errors" + "testing" + "time" + + healthDomain "proxy-pool/internal/domain/health" + proxyDomain "proxy-pool/internal/domain/proxy" +) + +func TestMemoryPoolKeepsTargetHealthOutsideGlobalProxyState(t *testing.T) { + now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) + pool := seededHealthPool(t, now) + before := pool.Snapshot(now) + if len(before) != 1 || before[0].State != proxyDomain.StateFetched { + t.Fatalf("before = %+v", before) + } + command := TargetHealthCommand{Observation: healthDomain.Observation{ + TaskID: "target-task-1", ProxyID: "proxy-a", Level: healthDomain.LevelTarget, + RoutingName: "route-a", TargetURL: "https://target.example/check", Success: false, + FailureClass: "target_403", Latency: 20 * time.Millisecond, ObservedAt: now.Add(time.Second), + }, MaxConsecutiveFailures: 2} + first, err := pool.ApplyTargetObservation(context.Background(), command) + if err != nil || first.Status != healthDomain.TargetSuspect || first.ConsecutiveFailures != 1 { + t.Fatalf("ApplyTargetObservation(first) = %+v, %v", first, err) + } + after := pool.Snapshot(now.Add(time.Second)) + if len(after) != 1 || after[0].State != before[0].State || after[0].GlobalHealth != before[0].GlobalHealth { + t.Fatalf("target observation changed global entry: before=%+v after=%+v", before[0], after[0]) + } + replayed, err := pool.ApplyTargetObservation(context.Background(), command) + if err != nil || replayed != first { + t.Fatalf("ApplyTargetObservation(replay) = %+v, %v", replayed, err) + } + conflicting := command + conflicting.Observation.Success = true + conflicting.Observation.FailureClass = "" + if _, err := pool.ApplyTargetObservation(context.Background(), conflicting); !errors.Is(err, healthDomain.ErrConflictingObservation) { + t.Fatalf("ApplyTargetObservation(conflict) error = %v, want ErrConflictingObservation", err) + } +}