diff --git a/README.md b/README.md index 2df10be..5a3f446 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 - **Gateway**：调用方连接平台，由平台选择上游代理并转发 HTTP 或 HTTPS CONNECT。`proxy-gateway` 已装配本地监听、指标探针和控制面 Register/Watch/ACK/ - Runtime 会话；带凭据 Proxy 分发仍待闭环。 + Runtime/Outcome 会话；带凭据 Proxy 的分发与当前内存 View 已闭环。 - **Distribution**：调用方按条件提取真实代理；成功提取即独占消费，不支持归还、 续租或状态查询。 - **Admin**：运维人员查询状态、启停 Upstream、切换 Routing，并触发严格配置 @@ -43,7 +43,9 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 审计与 Outbox；不保存 Proxy 明细或逐次提取记录。 - **Gateway 组件**：HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、重试、超时、 目的地址保护、本地快照存储、容量调度和 Worker 控制面 Register/Watch/ACK/ - Runtime 会话组件已有实现与定向测试。`SessionSupervisor` 会为可恢复控制面中断执行 + Runtime/Outcome 会话组件已有实现与定向测试。每次代理尝试只向本地有界队列写入 + Outcome，微批确认失败会重发同一序列；队列满时丢弃观测样本，不阻塞转发请求。 + `SessionSupervisor` 会为可恢复控制面中断执行 有界退避重连。Controller 可向 Worker 下发已归属 Proxy、Gateway Routing 与按引用去重的 凭据材料快照。Gateway 会将 Routing、Proxy 与凭据原子编译为同一内存 View，并只按当前未过期 View 匹配请求，并在内存中按 Sequential、Random、Round Robin、Weighted 或 Least @@ -78,18 +80,17 @@ flowchart LR ## 当前完成度 -截至 **2026-07-31**，实施计划检查项为 **55 / 74（74.3%）**。详情见 +截至 **2026-07-31**，实施计划检查项为 **56 / 74（75.7%）**。详情见 [实施计划](docs/development/implementation-plan.md)和 [交付完成度审计](docs/requirements/completion-audit.md)。 - **已完成**：严格配置、Provider 获取与协调、Redis 活动池、Distribution 原子 提取与限流、Controller 的 Admin/Distribution/Metrics 监听，以及 PostgreSQL 管理状态；WorkerControlPlane 的 Register、Snapshot ACK、Runtime 心跳接收和 - Redis 会话栅栏。 -- **部分完成**：Gateway Outcome 上报、Checker 调度与健康状态链、Docker Compose/ + Redis 会话栅栏，以及 Gateway Outcome 上报的有界队列、序列确认与重试。 +- **部分完成**：Checker 调度与健康状态链、Docker Compose/ Kubernetes 运行时 mTLS Overlay，以及 protobuf 契约。 -- **待完成**：Outcome 上报、Checker 调度与健康状态链， - 以及 loadgen 和代表性集群压测。 +- **待完成**：Checker 调度与健康状态链，以及 loadgen 和代表性集群压测。 检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表 端到端拓扑已经完成；`100,000 QPS` 仍只是待验证的集群设计目标。 diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index d38ad38..a1caebc 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -35,10 +35,13 @@ Gateway 已将该 payload 编译并原子发布到与 Proxy 相同版本的本 当前未过期 View；派发器已按五种策略从该 View 选择上游，且在 Proxy 容量耗尽时只在该 View 的其余候选中回退。`wait_timeout` 随 `on_unavailable=WAIT` 下发并在 Gateway 作为有界 本地容量等待使用；`DIRECT` 仍先经过 TargetPolicy 再建立 HTTP/CONNECT 直连。`proxy-gateway` -已装配 Register/Watch/ACK/Runtime 会话、HTTP 代理监听和 Snapshot 就绪探针；控制面中断时 -保持进程运行并以有界退避重连，未取得有效 Snapshot 的 Worker 不会 Ready。增量、带凭据 -Outcome 与 Checker 闭环尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`； -`100,000 QPS` 仍是未验证的设计目标。 +已装配 Register/Watch/ACK/Runtime/Outcome 会话、HTTP 代理监听和 Snapshot 就绪探针；控制面中断时 +保持进程运行并以有界退避重连，未取得有效 Snapshot 的 Worker 不会 Ready。`ReportOutcomes` +已实现为客户端流：Controller 校验流内固定的 Worker/Session 身份，对每个批次以 +`(session_id, sequence, SHA-256)` 建立 Redis 原子栅栏，并只返回最后确认的序列。 +Redis 只保留每个当前会话的一条序列和摘要；原始 Outcome、代理明细与逐请求记录均不写入 +Redis 或 PostgreSQL。Gateway 只将结果写入本地有界队列，队列满时丢弃样本，不等待控制面 +或存储。Checker 闭环尚未实现；`100,000 QPS` 仍是未验证的设计目标。 `WatchSnapshots` 建立时校验当前 session；每次签发快照引用时也把 `session_id` 交给 Redis 原子校验。重复 Register 会同时清除旧 Runtime 和已签发引用，因此迟到的 @@ -137,8 +140,15 @@ Outcome 按 Worker 单调 `sequence` 批量上报。Controller 返回已接受 - `TUNNEL`：隧道建立后结束或失败。 Outcome 是 Observation，不直接让 Worker 修改 Redis 活动池状态，也不创建 -PostgreSQL Proxy 明细。异步上报队列必须有界；队列满时丢弃低价值样本并计 -指标，不能反压 Gateway 热路径。 +PostgreSQL Proxy 明细。Gateway 将每次尝试写入进程内有界队列，当前默认容量为 +`65536` 个事件，单批最多为 `min(512, maxRuntimeCounters)`；一次 ReportOutcomes RPC +最多合并 16 个已就绪批次。队列满时只递增本地丢弃计数，不反压 Gateway 热路径。 + +同一 session 内，Controller 仅接受递增 `sequence`；相同序列且摘要一致视为幂等 +重放，相同序列且摘要不同返回 `AlreadyExists`，较小序列返回 `Aborted`。Gateway 在 +未收到确认时保留并重发完全相同的批次。暂态传输错误在当前 session 内按有上限的 +退避重试；会话栅栏错误交由 session supervisor 重建 session。栅栏随 session 替换 +或过期清理，因此长期在线 Worker 不会因独立 Outcome TTL 接受旧序列。 ## 6. Checker 任务 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index d00031c..e5bab38 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -190,6 +190,11 @@ Gateway 连接 Controller 时使用独立启动参数而非 `controlPlane.listen 模板保持 `controlPlane.enabled: false`，环境 Overlay 挂载 mTLS 证书并启用后才可启动 Gateway。 +`maxRuntimeCounters` 同时限制单个 Runtime 报告和单个 Outcome 批次的条目数。Gateway +在本地维护容量为 `65536` 的非阻塞 Outcome 队列，默认微批上限为 `512`，实际取二者中 +较小值；该队列与其序列确认状态仅存在于 Gateway 进程内。Controller 的 Redis 状态只保存 +当前 Worker session 的最后确认序列和摘要，PostgreSQL 不保存 Outcome、代理明细或逐请求记录。 + ## 5. Gateway ```yaml diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index ae2c866..bc3ddb9 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -223,7 +223,7 @@ payload 已按配置顺序和 Admin revision/current 状态发布并覆盖 check Proxy 原子编译为同版本 View，动态 Router 只匹配该未过期 View。派发器的五种上游选择已 接入该 View，并在容量耗尽时在同版本候选中回退；`onUnavailable` 的 reject、wait 与 direct 已接入 Gateway；`proxy-gateway` 已装配本地 HTTP/Metrics 监听、快照就绪探针和 -控制面重连与快照凭据分发，Outcome 上报仍未实现。 +控制面重连与快照凭据分发，以及有界 Outcome 上报、序列确认和精确重试。 已新增公用 `domain/activitypool` 契约及并发安全内存参考实现，Provider Reconciler 通过 `UpsertFetched` 写入带供应商 TTL 和分配安全余量的批次；已覆盖 @@ -261,7 +261,8 @@ Controller 多副本共享同一计数。Client 身份只以 SHA-256 摘要进 Gateway 请求热路径仍只使用本地准入，不增加 Redis/PostgreSQL 调用。 WorkerControlPlane gRPC 接收端、session 签发/心跳、Snapshot ACK 账本、基础 Snapshot 流、Gateway 会话客户端与快照凭据分发已完成；权威 Proxy/Routing 发布、 -Outcome 和健康执行链仍待完成，因此 Task 10 尚未全部完成。 +Outcome 上报已完成为有界队列、批次序列/摘要栅栏和确认重试；健康执行链仍待完成， +因此 Task 10 尚未全部完成。 ## Task 11: Checker and Health Reducer diff --git a/docs/operations/runbook.md b/docs/operations/runbook.md index 635bb26..47a92d0 100644 --- a/docs/operations/runbook.md +++ b/docs/operations/runbook.md @@ -3,6 +3,9 @@ ## 1. 运行边界 - Gateway 是数据面，正常请求热路径不访问 PostgreSQL、Redis 或 Provider。 +- Gateway 的 Proxy Outcome 仅进入进程内有界队列；Controller 只维护当前 session + 的最后确认序列和摘要，原始 Outcome 不写入 Redis 或 PostgreSQL。队列满时丢弃 + 观测样本，代理转发和 Redis session TTL 不受影响。 - Controller 编排 Fetch、生命周期、所有权、Snapshot 与 Extract；多个副本只有 一个 Provider 逻辑 Leader。短效 Proxy 明细只存在于 Redis TTL 活动池和节点 内存，可由 Provider 重建。 diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 2da6c4b..0cf5672 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -38,7 +38,7 @@ - `CAP-001 / GW 热路径边界`：本地 Dispatch 条件过滤与原子容量预留。 - `GW-*`：HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、超时、重试、保护链与 优雅停机 Handler 已实现并通过定向测试。`proxy-gateway` 已装配本地 HTTP/Metrics - 监听、快照就绪探针、显式控制面拨号和 Register/Watch/ACK/Runtime 会话；本地 gRPC + 监听、快照就绪探针、显式控制面拨号和 Register/Watch/ACK/Runtime/Outcome 会话；本地 gRPC 集成测试验证收到有效 Snapshot 后才 Ready，并可完成 direct 路由。 - `PROVIDER-*`：Provider HTTP Client、严格响应上限、模板解析安全边界、凭据 引用 Store 与 Reconciler Adapter 已实现。 @@ -56,7 +56,8 @@ 索引构建已归属 Proxy payload，并以租约收紧可用期；凭据材料按引用去重并与 Proxy/ Routing 一起纳入 Snapshot checksum，仅在 mTLS 和 Gateway 当前内存 View 中处理。Gateway Routing payload 已由配置顺序和 Admin 当前状态合成并覆盖 Snapshot checksum，其运行时消费、 -Outcome 和 Checker 尚未闭环。Snapshot 签发在 Redis 中原子匹配当前 +Outcome 已实现为 Gateway 本地有界队列、微批确认重试和 Controller 的 session/sequence/ +摘要 Redis 栅栏；原始事件不落 Redis 或 PostgreSQL。Checker 尚未闭环。Snapshot 签发在 Redis 中原子匹配当前 `session_id`，重注册会清除旧引用，迟到旧 Stream 不会覆盖新 session。Controller 在最近成功下发的 Snapshot `valid_until` 到达时关闭流；Gateway 的公用 `SessionSupervisor` 已实现可恢复错误的有界退避重连。Gateway 会校验并执行 Snapshot @@ -105,7 +106,7 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 分布式请求额度、Distribution Client 限制和 Provider 状态丢失重建已完成。 6. Worker 基础网络快照流、Proxy/Gateway Routing/凭据 Snapshot payload、Gateway Snapshot 客户端和进程装配、同版本 Routing 编译/动态匹配、五种策略上游选择与 reject/wait/direct 已完成； - Outcome 上报仍待完成。Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。 + Outcome 上报已完成基础观测链；Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。 7. Checker 调度、探测器和健康 reducer。 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index 3dd217e..a9d7576 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -7,8 +7,8 @@ | ID | 最终需求 | 来源 | 验证证据 | |---|---|---|---| -| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Gateway 已将它们编译为同版本动态 View，并由独立进程维护控制面会话。Checker/Loadgen 构建产物待实现 | -| ARCH-002 | 热路径只做认证、本地路由和网络转发 | 1-70, 380-430 | Gateway bootstrap 集成测试验证启动期控制面会话与快照就绪，HTTP 请求只走本地 Snapshot/Dispatch；代表性性能剖析待完成 | +| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime/Outcome gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Gateway 已将它们编译为同版本动态 View，并由独立进程维护控制面会话。Checker/Loadgen 构建产物待实现 | +| 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-005 | 100k QPS 峰值使用多 Worker 集群 | 当前会话 | 未验证设计目标；待代表性集群负载报告 | @@ -60,6 +60,7 @@ | GW-003 | CONNECT 建立后不得透明重放 | 221-300 | `handler_test.go` 200 前重试、200 后中继故障不重放、半关闭测试 | | GW-004 | Client 认证可关闭，但访问控制、身份识别和限流独立 | 8112-8441 | `protection_test.go`、`bootstrap_test.go` 与入口并发上限测试 | | GW-005 | 防私网、回环、链路本地、保留/元数据地址、任意 CONNECT 端口和 DNS Rebinding | 8904-8931 | `policy/target_test.go` 解析前后校验、端口白名单、混合 DNS 结果与已验证 IP 绑定测试 | +| GW-006 | Proxy Outcome 不反压转发并按会话序列去重 | 当前会话 | `gateway/outcome` 队列溢出与微批测试、`controlplane/outcome_reporter_test.go` 精确重试测试、Worker gRPC 和 Memory/Redis 共享栅栏契约 | ## Distribution diff --git a/docs/testing/test-strategy.md b/docs/testing/test-strategy.md index 9584db2..f034030 100644 --- a/docs/testing/test-strategy.md +++ b/docs/testing/test-strategy.md @@ -46,7 +46,7 @@ - Distribution 原子提取后 Gateway 不再分配同一 Proxy。 - 配置热更新失败保留旧 Revision，成功后新请求使用新 Snapshot。 -## 3. 必测的 11 类场景 +## 3. 必测的 12 类场景 1. **并发容量**：1000 协程争用同一 Proxy，始终满足 `active + reserved <= effectiveMaxConcurrency`。 @@ -68,6 +68,8 @@ unowned，旧 Snapshot 不可再分配。 11. **控制面故障**：Redis、PostgreSQL、Controller、Checker 与 Provider 分别 失效时，行为与 Runbook 一致，Gateway 热路径不被同步依赖拖垮。 +12. **Outcome 栅栏**：队列写入不阻塞转发；同一 session 的相同序列/摘要可重放， + 相同序列的不同摘要冲突，较小序列拒绝；Session 替换后旧 Worker 结果拒绝。 ## 4. 测试命令 diff --git a/internal/adapters/redisactivity/keys.go b/internal/adapters/redisactivity/keys.go index 5b7d976..ab602b5 100644 --- a/internal/adapters/redisactivity/keys.go +++ b/internal/adapters/redisactivity/keys.go @@ -26,6 +26,7 @@ type keyspace struct { workerSnapshotExpiry string workerRuntime string workerRuntimeExpiry string + workerOutcomes string } func newKeyspace(namespace string) keyspace { @@ -48,6 +49,7 @@ func newKeyspace(namespace string) keyspace { workerSnapshotExpiry: prefix + ":worker-snapshot-expiry", workerRuntime: prefix + ":worker-runtime", workerRuntimeExpiry: prefix + ":worker-runtime-expiry", + workerOutcomes: prefix + ":worker-outcomes", } } diff --git a/internal/adapters/redisactivity/runtime.go b/internal/adapters/redisactivity/runtime.go index c02cbc7..1e1ae37 100644 --- a/internal/adapters/redisactivity/runtime.go +++ b/internal/adapters/redisactivity/runtime.go @@ -20,6 +20,7 @@ const ( runtimeRecordSnapshot = "record_snapshot" runtimeAcknowledge = "acknowledge_snapshot" runtimeReplaceReport = "replace_report" + runtimeRecordOutcomes = "record_outcomes" runtimeRead = "read" ) @@ -84,6 +85,13 @@ type runtimeReportWire struct { Counters []runtimeCounterWire `json:"counters"` } +type runtimeOutcomeReferenceWire struct { + Version int `json:"version"` + WorkerID string `json:"workerId"` + SessionID string `json:"sessionId"` + Sequence string `json:"sequence"` +} + type runtimeOwnedProxyWire struct { ProxyID string `json:"proxyId"` WorkerID string `json:"workerId"` @@ -103,6 +111,7 @@ var ( _ workerruntime.SessionWriter = (*Adapter)(nil) _ workerruntime.ReportWriter = (*Adapter)(nil) _ workerruntime.RuntimeReader = (*Adapter)(nil) + _ workerruntime.OutcomeWriter = (*Adapter)(nil) ) func (a *Adapter) CurrentOwnershipEpoch(ctx context.Context) (uint64, error) { @@ -317,6 +326,48 @@ func (a *Adapter) ReplaceRuntime(ctx context.Context, report workerruntime.Repor } } +// RecordOutcomes persists only the monotonically increasing batch fence. The +// Gateway event payload is deliberately excluded from Redis to keep the data +// plane hot path bounded and free of per-request storage writes. +func (a *Adapter) RecordOutcomes(ctx context.Context, reference workerruntime.OutcomeReference) (uint64, error) { + if err := validateRuntimeCall(ctx, a); err != nil { + return 0, err + } + normalized, err := workerruntime.NormalizeOutcomeReference(reference) + if err != nil { + return 0, workerruntime.ErrInvalidOutcome + } + payload, err := json.Marshal(runtimeOutcomeReferenceWire{ + Version: runtimeWireVersion, WorkerID: normalized.WorkerID, SessionID: normalized.SessionID, + Sequence: strconv.FormatUint(normalized.Sequence, 10), + }) + if err != nil { + return 0, workerruntime.ErrInvalidOutcome + } + reply, err := a.runRuntime(ctx, runtimeRecordOutcomes, 0, payload, hex.EncodeToString(normalized.Digest[:])) + if err != nil { + return 0, err + } + switch reply.Status { + case scriptOK: + accepted, parseErr := strconv.ParseUint(reply.Record, 10, 64) + if parseErr != nil || accepted == 0 { + return 0, invalidScriptReply("invalid worker outcome reply") + } + return accepted, nil + case scriptInvalid: + return 0, workerruntime.ErrInvalidOutcome + case scriptStale: + return 0, workerruntime.ErrStaleOutcome + case scriptConflict: + return 0, workerruntime.ErrConflictingOutcome + case scriptUnavailable: + return 0, workerruntime.ErrStaleSession + default: + return 0, invalidScriptReply("unexpected worker outcome reply") + } +} + func (a *Adapter) ReadRuntime(ctx context.Context, proxies []workerruntime.OwnedProxy) ([]workerruntime.Snapshot, error) { if err := validateRuntimeCall(ctx, a); err != nil { return nil, err @@ -403,6 +454,7 @@ func (a *Adapter) runRuntime( a.keys.workerSessions, a.keys.workerSessionExpiry, a.keys.workerSnapshots, a.keys.workerSnapshotExpiry, a.keys.workerRuntime, a.keys.workerRuntimeExpiry, a.keys.owners, a.keys.epoch, + a.keys.workerOutcomes, }, operation, ttlMS, a.options.CleanupLimit, string(payload), digest) if err != nil { return runtimeScriptReply{}, err diff --git a/internal/adapters/redisactivity/runtime_contract_integration_test.go b/internal/adapters/redisactivity/runtime_contract_integration_test.go index 81878ba..3a02692 100644 --- a/internal/adapters/redisactivity/runtime_contract_integration_test.go +++ b/internal/adapters/redisactivity/runtime_contract_integration_test.go @@ -20,7 +20,7 @@ func TestRedisWorkerControlStoreContract(t *testing.T) { t.Fatalf("Assign(): %v", err) } return contracttest.Fixture{ - Store: fixture.Adapter, Reader: fixture.Adapter, TTL: 100 * time.Millisecond, + Store: fixture.Adapter, Reader: fixture.Adapter, Outcomes: fixture.Adapter, TTL: 100 * time.Millisecond, Advance: time.Sleep, } }) diff --git a/internal/adapters/redisactivity/scripts/runtime.lua b/internal/adapters/redisactivity/scripts/runtime.lua index c46fb5f..9c16d91 100644 --- a/internal/adapters/redisactivity/scripts/runtime.lua +++ b/internal/adapters/redisactivity/scripts/runtime.lua @@ -6,6 +6,7 @@ local runtime_key = KEYS[5] local runtime_expiry_key = KEYS[6] local owners_key = KEYS[7] local epoch_key = KEYS[8] +local outcomes_key = KEYS[9] local operation = ARGV[1] local ttl_ms = tonumber(ARGV[2]) @@ -60,6 +61,7 @@ local function cleanup(now) for _, worker_id in ipairs(expired_sessions) do redis.call('HDEL', sessions_key, worker_id) redis.call('HDEL', runtime_key, worker_id) + redis.call('HDEL', outcomes_key, worker_id) redis.call('ZREM', session_expiry_key, worker_id) redis.call('ZREM', runtime_expiry_key, worker_id) end @@ -106,6 +108,13 @@ local function valid_reference(value) string.match(value.checksum, '^[0-9a-f]+$') ~= nil end +local function valid_outcome(value) + return value and value.version == 1 and type(value.workerId) == 'string' and value.workerId ~= '' and + type(value.sessionId) == 'string' and value.sessionId ~= '' and valid_uint(value.sequence) and + type(value.digest) == 'string' and string.len(value.digest) == 64 and + string.match(value.digest, '^[0-9a-f]+$') ~= nil +end + local function compare_reference(left, right) local epoch_order = compare_uint(left.ownershipEpoch, right.ownershipEpoch) if epoch_order ~= 0 then @@ -148,6 +157,7 @@ if operation == 'open_session' then end redis.call('HDEL', runtime_key, session.workerId) redis.call('ZREM', runtime_expiry_key, session.workerId) + redis.call('HDEL', outcomes_key, session.workerId) redis.call('HDEL', snapshots_key, session.workerId) redis.call('ZREM', snapshot_expiry_key, session.workerId) session.expiresAtMs = now + ttl_ms @@ -301,10 +311,12 @@ if operation == 'replace_session' then if epoch_order > 0 or version_order > 0 then redis.call('HDEL', runtime_key, session.workerId) redis.call('ZREM', runtime_expiry_key, session.workerId) + redis.call('HDEL', outcomes_key, session.workerId) end else redis.call('HDEL', runtime_key, session.workerId) redis.call('ZREM', runtime_expiry_key, session.workerId) + redis.call('HDEL', outcomes_key, session.workerId) end session.expiresAtMs = now + ttl_ms redis.call('HSET', sessions_key, session.workerId, cjson.encode(session)) @@ -312,6 +324,39 @@ if operation == 'replace_session' then return reply('ok') end +if operation == 'record_outcomes' then + if type(digest) ~= 'string' or string.len(digest) ~= 64 or + string.match(digest, '^[0-9a-f]+$') == nil then + return reply('invalid') + end + local reference = decode_table(payload) + if not reference or reference.version ~= 1 or type(reference.workerId) ~= 'string' or reference.workerId == '' or + type(reference.sessionId) ~= 'string' or reference.sessionId == '' or not valid_uint(reference.sequence) then + return reply('invalid') + end + local session = decode_table(redis.call('HGET', sessions_key, reference.workerId)) + if not valid_session(session) or session.workerId ~= reference.workerId or session.sessionId ~= reference.sessionId or + type(session.expiresAtMs) ~= 'number' or session.expiresAtMs <= now then + return reply('unavailable') + end + local current = decode_table(redis.call('HGET', outcomes_key, reference.workerId)) + if valid_outcome(current) and current.workerId == reference.workerId and current.sessionId == reference.sessionId then + local ordering = compare_uint(reference.sequence, current.sequence) + if ordering < 0 then + return reply('stale') + end + if ordering == 0 then + if current.digest == digest then + return reply('ok', nil, current.sequence) + end + return reply('conflict') + end + end + reference.digest = digest + redis.call('HSET', outcomes_key, reference.workerId, cjson.encode(reference)) + return reply('ok', nil, reference.sequence) +end + if operation == 'replace_report' then if not ttl_ms or ttl_ms <= 0 or type(digest) ~= 'string' or digest == '' then return reply('invalid') diff --git a/internal/controller/worker/grpc_handler.go b/internal/controller/worker/grpc_handler.go index d102de5..e04dc29 100644 --- a/internal/controller/worker/grpc_handler.go +++ b/internal/controller/worker/grpc_handler.go @@ -4,9 +4,11 @@ import ( "context" "crypto/sha256" "errors" + "io" "time" controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/domain/outcome" "proxy-pool/internal/domain/workerruntime" "google.golang.org/grpc" @@ -197,6 +199,84 @@ func (handler *GRPCHandler) ReportRuntime(ctx context.Context, request *controlp }, nil } +func (handler *GRPCHandler) ReportOutcomes(stream grpc.ClientStreamingServer[controlplanev1.OutcomeBatch, controlplanev1.ReportOutcomesResponse]) error { + if stream == nil || handler == nil || handler.service == nil || handler.identity == nil { + return grpcError(ErrInvalidCommand) + } + var workerID, sessionID string + var accepted uint64 + batches := 0 + for { + request, err := stream.Recv() + if errors.Is(err, io.EOF) { + if workerID == "" { + return grpcError(ErrInvalidCommand) + } + return stream.SendAndClose(&controlplanev1.ReportOutcomesResponse{AcceptedThroughSequence: accepted}) + } + if err != nil { + return err + } + if batches == outcome.MaxBatchesPerStream { + return grpcError(ErrInvalidCommand) + } + batch, err := decodeOutcomeBatch(request) + if err != nil { + return grpcError(err) + } + if workerID == "" { + if err := handler.authorize(stream.Context(), batch.WorkerID); err != nil { + return err + } + workerID, sessionID = batch.WorkerID, batch.SessionID + } else if batch.WorkerID != workerID || batch.SessionID != sessionID { + return grpcError(ErrInvalidCommand) + } + accepted, err = handler.service.ReportOutcomes(stream.Context(), batch) + if err != nil { + return grpcError(err) + } + batches++ + } +} + +func decodeOutcomeBatch(request *controlplanev1.OutcomeBatch) (outcome.Batch, error) { + if request == nil { + return outcome.Batch{}, ErrInvalidCommand + } + events := make([]outcome.Event, len(request.GetOutcomes())) + for index, item := range request.GetOutcomes() { + if item == nil || item.GetLatency() == nil || item.GetLatency().CheckValid() != nil || + item.GetObservedAt() == nil || item.GetObservedAt().CheckValid() != nil { + return outcome.Batch{}, ErrInvalidCommand + } + stage, ok := decodeOutcomeStage(item.GetStage()) + if !ok { + return outcome.Batch{}, ErrInvalidCommand + } + events[index] = outcome.Event{ + ProxyID: item.GetProxyId(), RoutingName: item.GetRoutingName(), Stage: stage, Success: item.GetSuccess(), + ErrorClass: outcome.ErrorClass(item.GetErrorClass()), Latency: item.GetLatency().AsDuration(), ObservedAt: item.GetObservedAt().AsTime(), + } + } + return outcome.Batch{WorkerID: request.GetWorkerId(), SessionID: request.GetSessionId(), Sequence: request.GetSequence(), Events: events}, nil +} + +func decodeOutcomeStage(stage controlplanev1.OutcomeStage) (outcome.Stage, bool) { + switch stage { + case controlplanev1.OutcomeStage_OUTCOME_STAGE_DIAL: + return outcome.StageDial, true + case controlplanev1.OutcomeStage_OUTCOME_STAGE_PROXY_HANDSHAKE: + return outcome.StageProxyHandshake, true + case controlplanev1.OutcomeStage_OUTCOME_STAGE_RESPONSE_HEADERS: + return outcome.StageResponseHeaders, true + case controlplanev1.OutcomeStage_OUTCOME_STAGE_TUNNEL: + return outcome.StageTunnel, true + default: + return outcome.StageUnspecified, false + } +} + func (handler *GRPCHandler) authorize(ctx context.Context, workerID string) error { if err := handler.identity.Authorize(ctx, workerID); err != nil { return status.Error(codes.PermissionDenied, "worker identity is not authorized") @@ -210,7 +290,8 @@ func grpcError(err error) error { return status.Error(codes.Canceled, "worker control request canceled") case errors.Is(err, context.DeadlineExceeded): return status.Error(codes.DeadlineExceeded, "worker control request deadline exceeded") - case errors.Is(err, ErrInvalidCommand), errors.Is(err, workerruntime.ErrInvalidReport), + case errors.Is(err, ErrInvalidCommand), errors.Is(err, outcome.ErrInvalidBatch), errors.Is(err, workerruntime.ErrInvalidReport), + errors.Is(err, workerruntime.ErrInvalidOutcome), errors.Is(err, workerruntime.ErrInvalidAcknowledgement), errors.Is(err, workerruntime.ErrInvalidSnapshotReference): return status.Error(codes.InvalidArgument, "invalid worker control request") case errors.Is(err, ErrProtocolVersion): @@ -225,6 +306,10 @@ func grpcError(err error) error { return status.Error(codes.Aborted, "worker runtime sequence is stale") case errors.Is(err, workerruntime.ErrConflictingReport): return status.Error(codes.AlreadyExists, "worker runtime sequence conflicts") + case errors.Is(err, workerruntime.ErrStaleOutcome): + return status.Error(codes.Aborted, "worker outcome sequence is stale") + case errors.Is(err, workerruntime.ErrConflictingOutcome): + return status.Error(codes.AlreadyExists, "worker outcome sequence conflicts") default: return status.Error(codes.Unavailable, "worker control plane unavailable") } diff --git a/internal/controller/worker/grpc_handler_test.go b/internal/controller/worker/grpc_handler_test.go index 17427e8..84f6025 100644 --- a/internal/controller/worker/grpc_handler_test.go +++ b/internal/controller/worker/grpc_handler_test.go @@ -8,12 +8,15 @@ import ( "time" controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/domain/outcome" "proxy-pool/internal/domain/workerruntime" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/grpc/test/bufconn" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -45,7 +48,7 @@ func TestGRPCHandlerMapsWorkerRequests(t *testing.T) { } } -func TestGRPCHandlerMapsErrorsAndLeavesStreamsUnimplemented(t *testing.T) { +func TestGRPCHandlerMapsErrorsAndLeavesSnapshotStreamUnimplemented(t *testing.T) { service := &grpcServiceStub{registerErr: ErrProtocolVersion} client, cleanup := grpcWorkerClient(t, service, allowIdentity{}) defer cleanup() @@ -58,11 +61,88 @@ func TestGRPCHandlerMapsErrorsAndLeavesStreamsUnimplemented(t *testing.T) { if err != nil || status.Code(streamErr) != codes.Unimplemented { t.Fatalf("WatchSnapshots() = %v, %v", err, streamErr) } - outcomes, err := client.ReportOutcomes(context.Background()) - _, outcomesErr := outcomes.CloseAndRecv() - if err != nil || status.Code(outcomesErr) != codes.Unimplemented { - t.Fatalf("ReportOutcomes() = %v, %v", err, outcomesErr) +} + +func TestGRPCHandlerReportsOutcomeStream(t *testing.T) { + service := &grpcServiceStub{} + client, cleanup := grpcWorkerClient(t, service, allowIdentity{}) + defer cleanup() + stream, err := client.ReportOutcomes(context.Background()) + if err != nil { + t.Fatalf("ReportOutcomes() = %v", err) } + if err := stream.Send(&controlplanev1.OutcomeBatch{ + WorkerId: "worker-a", SessionId: "session-a", Sequence: 3, + Outcomes: []*controlplanev1.ProxyOutcome{{ + ProxyId: "proxy-a", RoutingName: "route-a", Stage: controlplanev1.OutcomeStage_OUTCOME_STAGE_DIAL, + Success: false, ErrorClass: "timeout", Latency: durationpb.New(25 * time.Millisecond), ObservedAt: timestamppb.New(time.Now()), + }}, + }); err != nil { + t.Fatalf("Send() = %v", err) + } + response, err := stream.CloseAndRecv() + if err != nil || response.GetAcceptedThroughSequence() != 3 || service.outcome.Sequence != 3 || + service.outcome.Events[0].ErrorClass != outcome.ErrorClassTimeout { + t.Fatalf("CloseAndRecv() = %+v, %v; outcome = %+v", response, err, service.outcome) + } +} + +func TestGRPCHandlerFencesOutcomesThroughWorkerService(t *testing.T) { + now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) + store, err := workerruntime.NewMemoryStore(func() time.Time { return now }) + if err != nil { + t.Fatalf("NewMemoryStore() = %v", err) + } + service, err := NewService(store, Options{ + ProtocolVersion: 1, HeartbeatInterval: time.Second, SessionTTL: 3 * time.Second, + MaxStaleAge: time.Second, MaxRuntimeCounters: 4, + SessionID: func() (string, error) { return "session-a", nil }, + }) + if err != nil { + t.Fatalf("NewService() = %v", err) + } + client, cleanup := grpcWorkerClient(t, service, allowIdentity{}) + defer cleanup() + registered, err := client.RegisterWorker(context.Background(), &controlplanev1.RegisterWorkerRequest{ + WorkerId: "worker-a", InstanceId: "instance-a", Zone: "zone-a", SupportedProtocolVersion: 1, + }) + if err != nil || registered.GetSessionId() != "session-a" { + t.Fatalf("RegisterWorker() = %+v, %v", registered, err) + } + batch := &controlplanev1.OutcomeBatch{ + WorkerId: "worker-a", SessionId: "session-a", Sequence: 1, + Outcomes: []*controlplanev1.ProxyOutcome{{ + ProxyId: "proxy-a", Stage: controlplanev1.OutcomeStage_OUTCOME_STAGE_DIAL, Success: true, + Latency: durationpb.New(time.Millisecond), ObservedAt: timestamppb.New(now), + }}, + } + if accepted, err := reportOutcomeBatch(context.Background(), client, batch); err != nil || accepted != 1 { + t.Fatalf("ReportOutcomes(first) = %d, %v", accepted, err) + } + if accepted, err := reportOutcomeBatch(context.Background(), client, batch); err != nil || accepted != 1 { + t.Fatalf("ReportOutcomes(replay) = %d, %v", accepted, err) + } + conflicting := proto.Clone(batch).(*controlplanev1.OutcomeBatch) + conflicting.Outcomes[0].Success = false + conflicting.Outcomes[0].ErrorClass = "dial" + if _, err := reportOutcomeBatch(context.Background(), client, conflicting); status.Code(err) != codes.AlreadyExists { + t.Fatalf("ReportOutcomes(conflict) code = %s, want AlreadyExists; error=%v", status.Code(err), err) + } +} + +func reportOutcomeBatch(ctx context.Context, client controlplanev1.WorkerControlPlaneClient, batch *controlplanev1.OutcomeBatch) (uint64, error) { + stream, err := client.ReportOutcomes(ctx) + if err != nil { + return 0, err + } + if err := stream.Send(batch); err != nil { + return 0, err + } + response, err := stream.CloseAndRecv() + if err != nil { + return 0, err + } + return response.GetAcceptedThroughSequence(), nil } func TestGRPCHandlerStreamsIssuedFullSnapshots(t *testing.T) { @@ -166,6 +246,9 @@ type grpcServiceStub struct { acknowledgeErr error report workerruntime.Report reportErr error + outcome outcome.Batch + outcomeAccepted uint64 + outcomeErr error issued workerruntime.SnapshotReference issuedSessionID string validateErr error @@ -195,6 +278,13 @@ func (stub *grpcServiceStub) ReportRuntime(_ context.Context, report workerrunti stub.report = report return RuntimeDecision{AcceptedOwnershipEpoch: 9}, stub.reportErr } +func (stub *grpcServiceStub) ReportOutcomes(_ context.Context, batch outcome.Batch) (uint64, error) { + stub.outcome = batch + if stub.outcomeAccepted == 0 { + stub.outcomeAccepted = batch.Sequence + } + return stub.outcomeAccepted, stub.outcomeErr +} type allowIdentity struct{} diff --git a/internal/controller/worker/service.go b/internal/controller/worker/service.go index adfdb7f..586ada6 100644 --- a/internal/controller/worker/service.go +++ b/internal/controller/worker/service.go @@ -8,6 +8,7 @@ import ( "errors" "time" + "proxy-pool/internal/domain/outcome" ownershipDomain "proxy-pool/internal/domain/ownership" "proxy-pool/internal/domain/workerruntime" platformCredentials "proxy-pool/internal/platform/credentials" @@ -71,6 +72,7 @@ type Service interface { IssueSnapshot(context.Context, string, workerruntime.SnapshotReference) error Acknowledge(context.Context, SnapshotAcknowledgement) error ReportRuntime(context.Context, workerruntime.Report) (RuntimeDecision, error) + ReportOutcomes(context.Context, outcome.Batch) (uint64, error) } func (service *service) CurrentOwnershipEpoch(ctx context.Context) (uint64, error) { @@ -119,6 +121,7 @@ func (service *service) IssueSnapshot(ctx context.Context, sessionID string, ref type service struct { store workerruntime.ControlStore + outcomes workerruntime.OutcomeWriter options Options snapshots SnapshotSource } @@ -133,6 +136,7 @@ func NewService(store workerruntime.ControlStore, options Options) (Service, err options.SessionID = randomSessionID } result := &service{store: store, options: options} + result.outcomes, _ = store.(workerruntime.OutcomeWriter) if options.SnapshotReader != nil { var routing []RoutingSource if options.RoutingSource != nil { @@ -252,6 +256,31 @@ func (service *service) ReportRuntime(ctx context.Context, report workerruntime. return RuntimeDecision{AcceptedOwnershipEpoch: epoch}, nil } +// ReportOutcomes records only a batch digest and sequence fence. Raw Gateway +// observations are intentionally not persisted by the control plane. +func (service *service) ReportOutcomes(ctx context.Context, batch outcome.Batch) (uint64, error) { + if ctx == nil { + return 0, ErrInvalidCommand + } + if err := ctx.Err(); err != nil { + return 0, err + } + normalized, digest, err := outcome.NormalizeBatch(batch, service.options.MaxRuntimeCounters) + if err != nil { + return 0, errors.Join(ErrInvalidCommand, err) + } + if service.outcomes == nil { + return 0, ErrUnavailable + } + accepted, err := service.outcomes.RecordOutcomes(ctx, workerruntime.OutcomeReference{ + WorkerID: normalized.WorkerID, SessionID: normalized.SessionID, Sequence: normalized.Sequence, Digest: digest, + }) + if err != nil { + return 0, classifyStoreError(err) + } + return accepted, nil +} + func randomSessionID() (string, error) { var value [16]byte if _, err := rand.Read(value[:]); err != nil { @@ -266,8 +295,10 @@ func classifyStoreError(err error) error { } if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, workerruntime.ErrInvalidSession) || errors.Is(err, workerruntime.ErrInvalidReport) || + errors.Is(err, workerruntime.ErrInvalidOutcome) || errors.Is(err, workerruntime.ErrInvalidAcknowledgement) || errors.Is(err, workerruntime.ErrStaleSession) || errors.Is(err, workerruntime.ErrStaleReport) || errors.Is(err, workerruntime.ErrConflictingReport) || + errors.Is(err, workerruntime.ErrStaleOutcome) || errors.Is(err, workerruntime.ErrConflictingOutcome) || errors.Is(err, workerruntime.ErrSnapshotMismatch) || errors.Is(err, workerruntime.ErrStaleAcknowledgement) { return err } diff --git a/internal/controller/worker/service_test.go b/internal/controller/worker/service_test.go index afdef66..abf851d 100644 --- a/internal/controller/worker/service_test.go +++ b/internal/controller/worker/service_test.go @@ -7,6 +7,7 @@ import ( "testing" "time" + "proxy-pool/internal/domain/outcome" "proxy-pool/internal/domain/workerruntime" ) @@ -51,6 +52,13 @@ func TestServiceRegistersAcknowledgesAndReportsRuntime(t *testing.T) { if err != nil || decision.RequireFullSnapshot || decision.AcceptedOwnershipEpoch != registered.OwnershipEpoch { t.Fatalf("ReportRuntime() = %+v, %v", decision, err) } + accepted, err := service.ReportOutcomes(context.Background(), outcome.Batch{ + WorkerID: "worker-a", SessionID: registered.SessionID, Sequence: 1, + Events: []outcome.Event{{ProxyID: "proxy-a", Stage: outcome.StageDial, Success: true, ObservedAt: now}}, + }) + if err != nil || accepted != 1 { + t.Fatalf("ReportOutcomes() = %d, %v", accepted, err) + } decision, err = service.ReportRuntime(context.Background(), workerruntime.Report{ WorkerID: "worker-a", SessionID: registered.SessionID, Sequence: 2, SnapshotVersion: 8, OwnershipEpoch: registered.OwnershipEpoch, ObservedAt: now, diff --git a/internal/domain/outcome/outcome.go b/internal/domain/outcome/outcome.go new file mode 100644 index 0000000..975c4a6 --- /dev/null +++ b/internal/domain/outcome/outcome.go @@ -0,0 +1,110 @@ +package outcome + +import ( + "crypto/sha256" + "encoding/json" + "errors" + "regexp" + "time" +) + +const ( + maximumIdentifierBytes = 128 + MaxBatchesPerStream = 16 +) + +var ( + ErrInvalidBatch = errors.New("invalid proxy outcome batch") + identifier = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$`) +) + +// Stage identifies the furthest proxy-processing boundary reached by a +// request. It is deliberately independent of a particular transport error. +type Stage uint8 + +const ( + StageUnspecified Stage = iota + StageDial + StageProxyHandshake + StageResponseHeaders + StageTunnel +) + +// ErrorClass is a bounded category suitable for aggregation and metrics. +// It must never contain raw connection, target, or credential details. +type ErrorClass string + +const ( + ErrorClassDial ErrorClass = "dial" + ErrorClassTimeout ErrorClass = "timeout" + ErrorClassProxyResponse ErrorClass = "proxy_response" + ErrorClassHandshake ErrorClass = "handshake" + ErrorClassRelay ErrorClass = "relay" + ErrorClassCanceled ErrorClass = "canceled" + ErrorClassInternal ErrorClass = "internal" +) + +type Event struct { + ProxyID string + RoutingName string + Stage Stage + Success bool + ErrorClass ErrorClass + Latency time.Duration + ObservedAt time.Time +} + +type Batch struct { + WorkerID string + SessionID string + Sequence uint64 + Events []Event +} + +// NormalizeBatch validates one bounded report, deep-copies it, and returns a +// digest used by the Controller's durable idempotency fence. +func NormalizeBatch(batch Batch, maximumEvents int) (Batch, [sha256.Size]byte, error) { + if !validIdentifier(batch.WorkerID) || !validIdentifier(batch.SessionID) || batch.Sequence == 0 || + maximumEvents <= 0 || len(batch.Events) == 0 || len(batch.Events) > maximumEvents { + return Batch{}, [sha256.Size]byte{}, ErrInvalidBatch + } + normalized := batch + normalized.Events = append([]Event(nil), batch.Events...) + for index := range normalized.Events { + event := &normalized.Events[index] + if !validIdentifier(event.ProxyID) || (event.RoutingName != "" && !validIdentifier(event.RoutingName)) || + !validStage(event.Stage) || event.Latency < 0 || event.ObservedAt.IsZero() || + (event.Success && event.ErrorClass != "") || (!event.Success && !validErrorClass(event.ErrorClass)) { + return Batch{}, [sha256.Size]byte{}, ErrInvalidBatch + } + event.ObservedAt = event.ObservedAt.UTC() + } + payload, err := json.Marshal(normalized) + if err != nil { + return Batch{}, [sha256.Size]byte{}, ErrInvalidBatch + } + return normalized, sha256.Sum256(payload), nil +} + +func validIdentifier(value string) bool { + return len(value) <= maximumIdentifierBytes && identifier.MatchString(value) +} + +func validStage(stage Stage) bool { + switch stage { + case StageDial, StageProxyHandshake, StageResponseHeaders, StageTunnel: + return true + default: + return false + } +} + +func validErrorClass(class ErrorClass) bool { + switch class { + case ErrorClassDial, ErrorClassTimeout, ErrorClassProxyResponse, ErrorClassHandshake, + ErrorClassRelay, ErrorClassCanceled, ErrorClassInternal: + return true + default: + return false + } +} diff --git a/internal/domain/outcome/outcome_test.go b/internal/domain/outcome/outcome_test.go new file mode 100644 index 0000000..fd2a785 --- /dev/null +++ b/internal/domain/outcome/outcome_test.go @@ -0,0 +1,52 @@ +package outcome + +import ( + "errors" + "testing" + "time" +) + +func TestNormalizeBatchCanonicalizesAndDigestsOutcomes(t *testing.T) { + observedAt := time.Date(2026, 7, 31, 12, 0, 0, 0, time.FixedZone("UTC+8", 8*60*60)) + normalized, digest, err := NormalizeBatch(Batch{ + WorkerID: "worker-a", SessionID: "session-a", Sequence: 7, + Events: []Event{{ + ProxyID: "proxy-a", RoutingName: "gateway-default", Stage: StageResponseHeaders, + Success: false, ErrorClass: ErrorClassTimeout, Latency: 25 * time.Millisecond, ObservedAt: observedAt, + }}, + }, 4) + if err != nil { + t.Fatalf("NormalizeBatch(): %v", err) + } + if normalized.Events[0].ObservedAt.Location() != time.UTC || digest == [32]byte{} { + t.Fatalf("normalized outcome = %+v, digest = %x", normalized.Events[0], digest) + } + _, repeated, err := NormalizeBatch(normalized, 4) + if err != nil || digest != repeated { + t.Fatalf("repeated NormalizeBatch() digest = %x, %v; want %x", repeated, err, digest) + } +} + +func TestNormalizeBatchRejectsUnboundedOrUnsafeInput(t *testing.T) { + now := time.Now() + valid := Event{ProxyID: "proxy-a", Stage: StageDial, Success: false, ErrorClass: ErrorClassDial, ObservedAt: now} + tests := []struct { + name string + batch Batch + limit int + }{ + {name: "missing worker", batch: Batch{SessionID: "session-a", Sequence: 1, Events: []Event{valid}}, limit: 1}, + {name: "zero sequence", batch: Batch{WorkerID: "worker-a", SessionID: "session-a", Events: []Event{valid}}, limit: 1}, + {name: "empty batch", batch: Batch{WorkerID: "worker-a", SessionID: "session-a", Sequence: 1}, limit: 1}, + {name: "exceeds limit", batch: Batch{WorkerID: "worker-a", SessionID: "session-a", Sequence: 1, Events: []Event{valid, valid}}, limit: 1}, + {name: "success with error", batch: Batch{WorkerID: "worker-a", SessionID: "session-a", Sequence: 1, Events: []Event{{ProxyID: "proxy-a", Stage: StageDial, Success: true, ErrorClass: ErrorClassDial, ObservedAt: now}}}, limit: 1}, + {name: "unknown error class", batch: Batch{WorkerID: "worker-a", SessionID: "session-a", Sequence: 1, Events: []Event{{ProxyID: "proxy-a", Stage: StageDial, ErrorClass: "raw network failure", ObservedAt: now}}}, limit: 1}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if _, _, err := NormalizeBatch(test.batch, test.limit); !errors.Is(err, ErrInvalidBatch) { + t.Fatalf("NormalizeBatch() error = %v, want ErrInvalidBatch", err) + } + }) + } +} diff --git a/internal/domain/workerruntime/contract_external_test.go b/internal/domain/workerruntime/contract_external_test.go index 6be0f62..8d8b5c6 100644 --- a/internal/domain/workerruntime/contract_external_test.go +++ b/internal/domain/workerruntime/contract_external_test.go @@ -17,7 +17,7 @@ func TestMemoryStoreContract(t *testing.T) { t.Fatalf("NewMemoryStore(): %v", err) } return contracttest.Fixture{ - Store: store, Reader: store, TTL: time.Minute, + Store: store, Reader: store, Outcomes: store, TTL: time.Minute, Advance: func(duration time.Duration) { now = now.Add(duration) }, } }) diff --git a/internal/domain/workerruntime/contracttest/contract.go b/internal/domain/workerruntime/contracttest/contract.go index 988ea14..e50cb4d 100644 --- a/internal/domain/workerruntime/contracttest/contract.go +++ b/internal/domain/workerruntime/contracttest/contract.go @@ -11,10 +11,11 @@ import ( ) type Fixture struct { - Store workerruntime.ControlStore - Reader workerruntime.RuntimeReader - TTL time.Duration - Advance func(time.Duration) + Store workerruntime.ControlStore + Reader workerruntime.RuntimeReader + Outcomes workerruntime.OutcomeWriter + TTL time.Duration + Advance func(time.Duration) } type Factory func(*testing.T) Fixture @@ -25,6 +26,8 @@ func Run(t *testing.T, factory Factory) { t.Run("acknowledged runtime lifecycle", func(t *testing.T) { runLifecycle(t, newFixture(t, factory)) }) t.Run("negative acknowledgement fences runtime", func(t *testing.T) { runNegativeAck(t, newFixture(t, factory)) }) t.Run("session replacement fences issued snapshots", func(t *testing.T) { runSessionReplacementFence(t, newFixture(t, factory)) }) + t.Run("outcome sequence fences retries", func(t *testing.T) { runOutcomeFence(t, newFixture(t, factory)) }) + t.Run("outcome invalid call is a store error", func(t *testing.T) { runOutcomeInvalidCall(t, newFixture(t, factory)) }) } func runLifecycle(t *testing.T, fixture Fixture) { @@ -117,10 +120,57 @@ func runSessionReplacementFence(t *testing.T, fixture Fixture) { } } +func runOutcomeFence(t *testing.T, fixture Fixture) { + t.Helper() + ctx := context.Background() + open(t, fixture.Store, fixture.TTL) + first := workerruntime.OutcomeReference{ + WorkerID: "worker-a", SessionID: "session-a", Sequence: 1, + Digest: sha256.Sum256([]byte("outcome-1")), + } + accepted, err := fixture.Outcomes.RecordOutcomes(ctx, first) + if err != nil || accepted != first.Sequence { + t.Fatalf("RecordOutcomes(first) = %d, %v; want 1, nil", accepted, err) + } + accepted, err = fixture.Outcomes.RecordOutcomes(ctx, first) + if err != nil || accepted != first.Sequence { + t.Fatalf("RecordOutcomes(replay) = %d, %v; want 1, nil", accepted, err) + } + conflicting := first + conflicting.Digest = sha256.Sum256([]byte("outcome-conflict")) + if _, err := fixture.Outcomes.RecordOutcomes(ctx, conflicting); !errors.Is(err, workerruntime.ErrConflictingOutcome) { + t.Fatalf("RecordOutcomes(conflict) = %v, want ErrConflictingOutcome", err) + } + next := first + next.Sequence = 2 + next.Digest = sha256.Sum256([]byte("outcome-2")) + if accepted, err := fixture.Outcomes.RecordOutcomes(ctx, next); err != nil || accepted != next.Sequence { + t.Fatalf("RecordOutcomes(next) = %d, %v; want 2, nil", accepted, err) + } + if _, err := fixture.Outcomes.RecordOutcomes(ctx, first); !errors.Is(err, workerruntime.ErrStaleOutcome) { + t.Fatalf("RecordOutcomes(stale) = %v, want ErrStaleOutcome", err) + } + if err := fixture.Store.OpenSession(ctx, workerruntime.Session{ + WorkerID: "worker-a", InstanceID: "instance-b", SessionID: "session-b", Zone: "zone-a", ProtocolVersion: 1, + }, fixture.TTL); err != nil { + t.Fatalf("OpenSession(replacement): %v", err) + } + if _, err := fixture.Outcomes.RecordOutcomes(ctx, first); !errors.Is(err, workerruntime.ErrStaleSession) { + t.Fatalf("RecordOutcomes(old session) = %v, want ErrStaleSession", err) + } +} + +func runOutcomeInvalidCall(t *testing.T, fixture Fixture) { + t.Helper() + if _, err := fixture.Outcomes.RecordOutcomes(nil, workerruntime.OutcomeReference{}); !errors.Is(err, workerruntime.ErrInvalidStore) { + t.Fatalf("RecordOutcomes(nil) = %v, want ErrInvalidStore", err) + } +} + func newFixture(t *testing.T, factory Factory) Fixture { t.Helper() fixture := factory(t) - if fixture.Store == nil || fixture.Reader == nil || fixture.TTL <= 0 || fixture.Advance == nil { + if fixture.Store == nil || fixture.Reader == nil || fixture.Outcomes == nil || fixture.TTL <= 0 || fixture.Advance == nil { t.Fatal("contract fixture is incomplete") } return fixture diff --git a/internal/domain/workerruntime/memory.go b/internal/domain/workerruntime/memory.go index 6bb09bd..6811df7 100644 --- a/internal/domain/workerruntime/memory.go +++ b/internal/domain/workerruntime/memory.go @@ -14,6 +14,7 @@ type MemoryStore struct { sessions map[string]memorySession references map[string]memoryReference reports map[string]memoryReport + outcomes map[string]memoryOutcome } type memorySession struct { @@ -33,11 +34,16 @@ type memoryReport struct { counters map[string]Counter } +type memoryOutcome struct { + value OutcomeReference +} + var ( _ ControlStore = (*MemoryStore)(nil) _ SessionWriter = (*MemoryStore)(nil) _ ReportWriter = (*MemoryStore)(nil) _ RuntimeReader = (*MemoryStore)(nil) + _ OutcomeWriter = (*MemoryStore)(nil) ) func NewMemoryStore(now func() time.Time) (*MemoryStore, error) { @@ -46,7 +52,7 @@ func NewMemoryStore(now func() time.Time) (*MemoryStore, error) { } return &MemoryStore{ now: now, epoch: 1, sessions: make(map[string]memorySession), - references: make(map[string]memoryReference), reports: make(map[string]memoryReport), + references: make(map[string]memoryReference), reports: make(map[string]memoryReport), outcomes: make(map[string]memoryOutcome), }, nil } @@ -85,6 +91,7 @@ func (store *MemoryStore) OpenSession(ctx context.Context, session Session, ttl defer store.mu.Unlock() delete(store.reports, normalized.WorkerID) delete(store.references, normalized.WorkerID) + delete(store.outcomes, normalized.WorkerID) store.sessions[normalized.WorkerID] = memorySession{value: normalized, expiresAt: now.Add(ttl)} return nil } @@ -246,12 +253,49 @@ func (store *MemoryStore) ReplaceSession(ctx context.Context, session Session, t ackAdvanced := exists && !identityChanged && !expired && legacySessionAfter(session, current.value) if identityChanged || expired || ackAdvanced { delete(store.reports, session.WorkerID) + delete(store.outcomes, session.WorkerID) } session.RuntimeEnabled = true store.sessions[session.WorkerID] = memorySession{value: session, expiresAt: now.Add(ttl)} return nil } +func (store *MemoryStore) RecordOutcomes(ctx context.Context, reference OutcomeReference) (uint64, error) { + if ctx == nil || store == nil { + return 0, ErrInvalidStore + } + if err := ctx.Err(); err != nil { + return 0, err + } + normalized, err := NormalizeOutcomeReference(reference) + if err != nil { + return 0, err + } + now, err := store.currentTime() + if err != nil { + return 0, err + } + store.mu.Lock() + defer store.mu.Unlock() + session, exists := store.sessions[normalized.WorkerID] + if !exists || !session.expiresAt.After(now) || session.value.SessionID != normalized.SessionID { + return 0, ErrStaleSession + } + current, exists := store.outcomes[normalized.WorkerID] + if exists && current.value.SessionID == normalized.SessionID { + switch { + case normalized.Sequence < current.value.Sequence: + return 0, ErrStaleOutcome + case normalized.Sequence == current.value.Sequence && normalized.Digest != current.value.Digest: + return 0, ErrConflictingOutcome + case normalized.Sequence == current.value.Sequence: + return current.value.Sequence, nil + } + } + store.outcomes[normalized.WorkerID] = memoryOutcome{value: normalized} + return normalized.Sequence, nil +} + func (store *MemoryStore) ReplaceRuntime(ctx context.Context, report Report, ttl time.Duration) error { if ctx == nil || store == nil || ttl <= 0 { return ErrInvalidReport diff --git a/internal/domain/workerruntime/outcome_test.go b/internal/domain/workerruntime/outcome_test.go new file mode 100644 index 0000000..200f77d --- /dev/null +++ b/internal/domain/workerruntime/outcome_test.go @@ -0,0 +1,51 @@ +package workerruntime + +import ( + "context" + "crypto/sha256" + "errors" + "testing" + "time" +) + +func TestMemoryStoreFencesOutcomeSequenceBySessionAndDigest(t *testing.T) { + now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) + store := newRuntimeStore(t, &now) + openControlSession(t, store, "session-a", time.Minute) + first := OutcomeReference{ + WorkerID: "worker-a", SessionID: "session-a", Sequence: 1, Digest: sha256.Sum256([]byte("batch-1")), + } + accepted, err := store.RecordOutcomes(context.Background(), first) + if err != nil || accepted != 1 { + t.Fatalf("RecordOutcomes(first) = %d, %v; want 1, nil", accepted, err) + } + accepted, err = store.RecordOutcomes(context.Background(), first) + if err != nil || accepted != 1 { + t.Fatalf("RecordOutcomes(replay) = %d, %v; want 1, nil", accepted, err) + } + conflicting := first + conflicting.Digest = sha256.Sum256([]byte("different")) + if _, err := store.RecordOutcomes(context.Background(), conflicting); !errors.Is(err, ErrConflictingOutcome) { + t.Fatalf("RecordOutcomes(conflicting replay) error = %v, want ErrConflictingOutcome", err) + } + if _, err := store.RecordOutcomes(context.Background(), OutcomeReference{ + WorkerID: "worker-a", SessionID: "session-a", Sequence: 0, Digest: first.Digest, + }); !errors.Is(err, ErrInvalidOutcome) { + t.Fatalf("RecordOutcomes(zero sequence) error = %v, want ErrInvalidOutcome", err) + } + if _, err := store.RecordOutcomes(context.Background(), OutcomeReference{ + WorkerID: "worker-a", SessionID: "session-a", Sequence: 2, Digest: sha256.Sum256([]byte("batch-2")), + }); err != nil { + t.Fatalf("RecordOutcomes(next) error = %v", err) + } + if _, err := store.RecordOutcomes(context.Background(), OutcomeReference{ + WorkerID: "worker-a", SessionID: "session-a", Sequence: 1, Digest: first.Digest, + }); !errors.Is(err, ErrStaleOutcome) { + t.Fatalf("RecordOutcomes(stale) error = %v, want ErrStaleOutcome", err) + } + + openControlSession(t, store, "session-b", time.Minute) + if _, err := store.RecordOutcomes(context.Background(), first); !errors.Is(err, ErrStaleSession) { + t.Fatalf("RecordOutcomes(old session) error = %v, want ErrStaleSession", err) + } +} diff --git a/internal/domain/workerruntime/runtime.go b/internal/domain/workerruntime/runtime.go index 15cfe01..929786f 100644 --- a/internal/domain/workerruntime/runtime.go +++ b/internal/domain/workerruntime/runtime.go @@ -21,6 +21,9 @@ var ( ErrStaleSnapshotReference = errors.New("stale worker snapshot reference") ErrConflictingSnapshotReference = errors.New("conflicting worker snapshot reference") ErrStaleAcknowledgement = errors.New("stale worker snapshot acknowledgement") + ErrInvalidOutcome = errors.New("invalid worker outcome reference") + ErrStaleOutcome = errors.New("stale worker outcome sequence") + ErrConflictingOutcome = errors.New("conflicting worker outcome sequence") ) type Session struct { @@ -84,6 +87,15 @@ type Snapshot struct { Fresh bool } +// OutcomeReference is the durable deduplication fence for one Gateway Outcome +// batch. The event payload itself remains outside the runtime store. +type OutcomeReference struct { + WorkerID string + SessionID string + Sequence uint64 + Digest [sha256.Size]byte +} + type SessionWriter interface { ReplaceSession(context.Context, Session, time.Duration) error } @@ -97,6 +109,10 @@ type ControlStore interface { ReplaceRuntime(context.Context, Report, time.Duration) error } +type OutcomeWriter interface { + RecordOutcomes(context.Context, OutcomeReference) (uint64, error) +} + type ReportWriter interface { ReplaceRuntime(context.Context, Report, time.Duration) error } diff --git a/internal/domain/workerruntime/validation.go b/internal/domain/workerruntime/validation.go index 0ea62b6..d058608 100644 --- a/internal/domain/workerruntime/validation.go +++ b/internal/domain/workerruntime/validation.go @@ -82,6 +82,14 @@ func NormalizeAcknowledgement(acknowledgement SnapshotAcknowledgement) (Snapshot return acknowledgement, nil } +func NormalizeOutcomeReference(reference OutcomeReference) (OutcomeReference, error) { + if !ValidIdentifier(reference.WorkerID) || !ValidIdentifier(reference.SessionID) || reference.Sequence == 0 || + checksumIsZero(reference.Digest) { + return OutcomeReference{}, ErrInvalidOutcome + } + return reference, nil +} + // NormalizeReport returns the canonical sparse replacement and its digest. func NormalizeReport(report Report) (Report, [sha256.Size]byte, error) { if !ValidIdentifier(report.WorkerID) || !ValidIdentifier(report.SessionID) || report.Sequence == 0 || diff --git a/internal/gateway/bootstrap/bootstrap.go b/internal/gateway/bootstrap/bootstrap.go index 351a4e6..5ebe4f0 100644 --- a/internal/gateway/bootstrap/bootstrap.go +++ b/internal/gateway/bootstrap/bootstrap.go @@ -27,6 +27,7 @@ import ( "proxy-pool/internal/domain/workerruntime" "proxy-pool/internal/gateway/controlplane" "proxy-pool/internal/gateway/dispatch" + gatewayOutcome "proxy-pool/internal/gateway/outcome" "proxy-pool/internal/gateway/server" "proxy-pool/internal/gateway/snapshot" "proxy-pool/internal/gateway/transport" @@ -45,6 +46,8 @@ const ( defaultReconnectInitialDelay = time.Second defaultReconnectMaxDelay = 30 * time.Second defaultReconnectJitter = 20 + defaultOutcomeQueueCapacity = 65_536 + defaultOutcomeBatchSize = 512 ) // Options provides process-local settings. ControlPlaneAddress is deliberately @@ -147,6 +150,13 @@ func newRuntime(ctx context.Context, configuration *config.Config, options Optio return nil, fmt.Errorf("build gateway target policy: %w", err) } proxyTransport := transport.New(transport.Config{}, snapshotCredentialResolver{store: store}) + outcomes, err := gatewayOutcome.NewQueue(gatewayOutcome.QueueOptions{ + Capacity: defaultOutcomeQueueCapacity, MaxBatch: min(defaultOutcomeBatchSize, configuration.ControlPlane.MaxRuntimeCounters), + }) + if err != nil { + proxyTransport.CloseIdleConnections() + return nil, fmt.Errorf("build gateway outcome queue: %w", err) + } handler, err := server.New(server.ConfigFromListener(configuration.Gateway), server.Dependencies{ Auth: protection.Auth, Access: protection.Access, @@ -155,6 +165,7 @@ func newRuntime(ctx context.Context, configuration *config.Config, options Optio Router: server.NewSnapshotRouter(store), Dispatcher: dispatch.New(store), Transport: proxyTransport, + Outcomes: outcomes, }) if err != nil { proxyTransport.CloseIdleConnections() @@ -198,7 +209,12 @@ func newRuntime(ctx context.Context, configuration *config.Config, options Optio if err != nil { return nil, fmt.Errorf("build snapshot watcher: %w", err) } - session, err := controlplane.NewSessionRunner(reporter, watcher) + outcomeReporter, err := controlplane.NewOutcomeReporter(controlplane.NewGeneratedOutcomeRPCClient(client), outcomes, + controlplane.OutcomeReporterOptions{WorkerID: options.WorkerID}) + if err != nil { + return nil, fmt.Errorf("build outcome reporter: %w", err) + } + session, err := controlplane.NewSessionRunner(reporter, watcher, outcomeReporter) if err != nil { return nil, fmt.Errorf("build control plane session: %w", err) } diff --git a/internal/gateway/controlplane/outcome_reporter.go b/internal/gateway/controlplane/outcome_reporter.go new file mode 100644 index 0000000..be5f1ae --- /dev/null +++ b/internal/gateway/controlplane/outcome_reporter.go @@ -0,0 +1,254 @@ +package controlplane + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + domain "proxy-pool/internal/domain/outcome" + "proxy-pool/internal/domain/workerruntime" + gatewayOutcome "proxy-pool/internal/gateway/outcome" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/timestamppb" +) + +var ( + ErrInvalidOutcomeReporter = errors.New("invalid gateway outcome reporter") + ErrOutcomeUnconfirmed = errors.New("controller did not confirm gateway outcome batch") +) + +const ( + outcomeRetryInitialDelay = 100 * time.Millisecond + outcomeRetryMaxDelay = 5 * time.Second +) + +type OutcomeStream interface { + Send(*controlplanev1.OutcomeBatch) error + CloseAndRecv() (*controlplanev1.ReportOutcomesResponse, error) +} + +type OutcomeRPCClient interface { + ReportOutcomes(context.Context) (OutcomeStream, error) +} + +type OutcomeReporterOptions struct { + WorkerID string + MaxBatchesPerReportRPC int +} + +type OutcomeReporter struct { + client OutcomeRPCClient + queue *gatewayOutcome.Queue + options OutcomeReporterOptions + + mu sync.Mutex + sessionID string + sequence uint64 + pending []domain.Batch +} + +func NewOutcomeReporter(client OutcomeRPCClient, queue *gatewayOutcome.Queue, options OutcomeReporterOptions) (*OutcomeReporter, error) { + if client == nil || queue == nil || !workerruntime.ValidIdentifier(options.WorkerID) { + return nil, ErrInvalidOutcomeReporter + } + if options.MaxBatchesPerReportRPC <= 0 { + options.MaxBatchesPerReportRPC = domain.MaxBatchesPerStream + } + if options.MaxBatchesPerReportRPC > domain.MaxBatchesPerStream { + return nil, ErrInvalidOutcomeReporter + } + return &OutcomeReporter{client: client, queue: queue, options: options}, nil +} + +// RunRegistered continuously batches local observations for a single +// Controller session. Transient delivery failures retain the exact pending +// batches and retry them in the same session with bounded backoff. +func (reporter *OutcomeReporter) RunRegistered(ctx context.Context, registration Registration) error { + if reporter == nil || ctx == nil || !workerruntime.ValidIdentifier(registration.SessionID) { + return ErrInvalidOutcomeReporter + } + reporter.mu.Lock() + reporter.sessionID = registration.SessionID + reporter.sequence = 0 + reporter.pending = nil + reporter.mu.Unlock() + retryDelay := outcomeRetryInitialDelay + for { + err := reporter.Report(ctx, registration.SessionID) + if err == nil { + retryDelay = outcomeRetryInitialDelay + continue + } + if ctx.Err() != nil { + return ctx.Err() + } + if terminalOutcomeError(err) { + return err + } + if err := waitOutcomeRetry(ctx, retryDelay); err != nil { + return err + } + retryDelay = nextOutcomeRetryDelay(retryDelay) + } +} + +func terminalOutcomeError(err error) bool { + if errors.Is(err, ErrInvalidOutcomeReporter) || errors.Is(err, ErrNotRegistered) { + return true + } + switch status.Code(err) { + case codes.Aborted, codes.AlreadyExists, codes.FailedPrecondition, + codes.InvalidArgument, codes.PermissionDenied, codes.Unauthenticated: + return true + default: + return false + } +} + +func waitOutcomeRetry(ctx context.Context, delay time.Duration) error { + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + +func nextOutcomeRetryDelay(delay time.Duration) time.Duration { + if delay >= outcomeRetryMaxDelay/2 { + return outcomeRetryMaxDelay + } + return delay * 2 +} + +func (reporter *OutcomeReporter) Report(ctx context.Context, sessionID string) error { + if reporter == nil || ctx == nil || !workerruntime.ValidIdentifier(sessionID) { + return ErrInvalidOutcomeReporter + } + if err := ctx.Err(); err != nil { + return err + } + reporter.mu.Lock() + defer reporter.mu.Unlock() + if reporter.sessionID != sessionID { + return ErrNotRegistered + } + if err := reporter.fillPending(ctx); err != nil { + return err + } + streamCtx, cancel := context.WithCancel(ctx) + defer cancel() + stream, err := reporter.client.ReportOutcomes(streamCtx) + if err != nil { + return err + } + for _, batch := range reporter.pending { + request, requestErr := outcomeRequest(batch) + if requestErr != nil { + return requestErr + } + if sendErr := stream.Send(request); sendErr != nil { + return sendErr + } + } + response, err := stream.CloseAndRecv() + if err != nil { + return err + } + last := reporter.pending[len(reporter.pending)-1].Sequence + if response == nil || response.GetAcceptedThroughSequence() != last { + return ErrOutcomeUnconfirmed + } + reporter.sequence = last + reporter.pending = nil + return nil +} + +func (reporter *OutcomeReporter) fillPending(ctx context.Context) error { + if len(reporter.pending) == 0 { + events, err := reporter.queue.Next(ctx) + if err != nil { + return err + } + if err := reporter.append(events); err != nil { + return err + } + } + for len(reporter.pending) < reporter.options.MaxBatchesPerReportRPC { + events, ok := reporter.queue.TryNext() + if !ok { + return nil + } + if err := reporter.append(events); err != nil { + return err + } + } + return nil +} + +func (reporter *OutcomeReporter) append(events []domain.Event) error { + batch, _, err := domain.NormalizeBatch(domain.Batch{ + WorkerID: reporter.options.WorkerID, SessionID: reporter.sessionID, + Sequence: reporter.sequence + uint64(len(reporter.pending)) + 1, Events: events, + }, reporter.queue.MaxBatch()) + if err != nil { + return errors.Join(ErrInvalidOutcomeReporter, err) + } + reporter.pending = append(reporter.pending, batch) + return nil +} + +func outcomeRequest(batch domain.Batch) (*controlplanev1.OutcomeBatch, error) { + proxies := make([]*controlplanev1.ProxyOutcome, len(batch.Events)) + for index, event := range batch.Events { + stage, ok := outcomeStage(event.Stage) + if !ok { + return nil, fmt.Errorf("%w: unknown outcome stage", ErrInvalidOutcomeReporter) + } + proxies[index] = &controlplanev1.ProxyOutcome{ + ProxyId: event.ProxyID, RoutingName: event.RoutingName, Stage: stage, Success: event.Success, + ErrorClass: string(event.ErrorClass), Latency: durationpb.New(event.Latency), ObservedAt: timestamppb.New(event.ObservedAt), + } + } + return &controlplanev1.OutcomeBatch{ + WorkerId: batch.WorkerID, SessionId: batch.SessionID, Sequence: batch.Sequence, Outcomes: proxies, + }, nil +} + +func outcomeStage(stage domain.Stage) (controlplanev1.OutcomeStage, bool) { + switch stage { + case domain.StageDial: + return controlplanev1.OutcomeStage_OUTCOME_STAGE_DIAL, true + case domain.StageProxyHandshake: + return controlplanev1.OutcomeStage_OUTCOME_STAGE_PROXY_HANDSHAKE, true + case domain.StageResponseHeaders: + return controlplanev1.OutcomeStage_OUTCOME_STAGE_RESPONSE_HEADERS, true + case domain.StageTunnel: + return controlplanev1.OutcomeStage_OUTCOME_STAGE_TUNNEL, true + default: + return controlplanev1.OutcomeStage_OUTCOME_STAGE_UNSPECIFIED, false + } +} + +type generatedOutcomeRPCClient struct { + client controlplanev1.WorkerControlPlaneClient +} + +func NewGeneratedOutcomeRPCClient(client controlplanev1.WorkerControlPlaneClient) OutcomeRPCClient { + return generatedOutcomeRPCClient{client: client} +} + +func (client generatedOutcomeRPCClient) ReportOutcomes(ctx context.Context) (OutcomeStream, error) { + if client.client == nil { + return nil, ErrInvalidOutcomeReporter + } + return client.client.ReportOutcomes(ctx) +} diff --git a/internal/gateway/controlplane/outcome_reporter_test.go b/internal/gateway/controlplane/outcome_reporter_test.go new file mode 100644 index 0000000..7d86e9e --- /dev/null +++ b/internal/gateway/controlplane/outcome_reporter_test.go @@ -0,0 +1,175 @@ +package controlplane + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + domain "proxy-pool/internal/domain/outcome" + gatewayOutcome "proxy-pool/internal/gateway/outcome" +) + +func TestOutcomeReporterRetriesExactPendingBatchAfterTransportFailure(t *testing.T) { + queue, err := gatewayOutcome.NewQueue(gatewayOutcome.QueueOptions{Capacity: 4, MaxBatch: 2}) + if err != nil { + t.Fatalf("NewQueue() = %v", err) + } + queue.Record(domain.Event{ProxyID: "proxy-a", Stage: domain.StageDial, Success: true, ObservedAt: time.Now()}) + client := &outcomeClientStub{closeErr: errors.New("connection dropped")} + reporter, err := NewOutcomeReporter(client, queue, OutcomeReporterOptions{WorkerID: "worker-a"}) + if err != nil { + t.Fatalf("NewOutcomeReporter() = %v", err) + } + registration := Registration{SessionID: "session-a"} + reporter.sessionID = registration.SessionID + if err := reporter.Report(context.Background(), registration.SessionID); !errors.Is(err, client.closeErr) { + t.Fatalf("Report(first) = %v, want transport failure", err) + } + client.closeErr = nil + if err := reporter.Report(context.Background(), registration.SessionID); err != nil { + t.Fatalf("Report(retry) = %v", err) + } + if len(client.batches) != 2 || client.batches[0].GetSequence() != 1 || client.batches[1].GetSequence() != 1 || + client.batches[0].GetOutcomes()[0].GetProxyId() != "proxy-a" { + t.Fatalf("sent batches = %+v", client.batches) + } +} + +func TestOutcomeReporterCombinesAvailableBatchesAndRequiresConfirmation(t *testing.T) { + queue, err := gatewayOutcome.NewQueue(gatewayOutcome.QueueOptions{Capacity: 8, MaxBatch: 1}) + if err != nil { + t.Fatalf("NewQueue() = %v", err) + } + for _, proxyID := range []string{"proxy-a", "proxy-b"} { + queue.Record(domain.Event{ProxyID: proxyID, Stage: domain.StageDial, Success: true, ObservedAt: time.Now()}) + } + client := &outcomeClientStub{} + reporter, err := NewOutcomeReporter(client, queue, OutcomeReporterOptions{WorkerID: "worker-a", MaxBatchesPerReportRPC: 2}) + if err != nil { + t.Fatalf("NewOutcomeReporter() = %v", err) + } + reporter.sessionID = "session-a" + if err := reporter.Report(context.Background(), "session-a"); err != nil { + t.Fatalf("Report() = %v", err) + } + if len(client.batches) != 2 || client.batches[0].GetSequence() != 1 || client.batches[1].GetSequence() != 2 { + t.Fatalf("sent batches = %+v", client.batches) + } +} + +func TestOutcomeReporterRetriesTransientFailureWithinRegisteredSession(t *testing.T) { + queue, err := gatewayOutcome.NewQueue(gatewayOutcome.QueueOptions{Capacity: 4, MaxBatch: 2}) + if err != nil { + t.Fatalf("NewQueue() = %v", err) + } + queue.Record(domain.Event{ProxyID: "proxy-a", Stage: domain.StageDial, Success: true, ObservedAt: time.Now()}) + client := &retryingOutcomeClient{failures: 1, confirmed: make(chan struct{}, 1)} + reporter, err := NewOutcomeReporter(client, queue, OutcomeReporterOptions{WorkerID: "worker-a"}) + if err != nil { + t.Fatalf("NewOutcomeReporter() = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- reporter.RunRegistered(ctx, Registration{SessionID: "session-a"}) }() + + select { + case <-client.confirmed: + case <-time.After(time.Second): + t.Fatal("outcome batch was not confirmed after transient failure") + } + if batches := client.Batches(); len(batches) != 2 || batches[0].GetSequence() != 1 || + batches[1].GetSequence() != 1 || batches[0].GetOutcomes()[0].GetProxyId() != "proxy-a" { + t.Fatalf("sent batches = %+v, want exact retry in the same session", batches) + } + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("RunRegistered() = %v, want context canceled", err) + } +} + +func TestOutcomeReporterRejectsUnboundedStreamOptions(t *testing.T) { + queue, err := gatewayOutcome.NewQueue(gatewayOutcome.QueueOptions{Capacity: 1, MaxBatch: 1}) + if err != nil { + t.Fatalf("NewQueue() = %v", err) + } + if _, err := NewOutcomeReporter(&outcomeClientStub{}, queue, OutcomeReporterOptions{ + WorkerID: "worker-a", MaxBatchesPerReportRPC: domain.MaxBatchesPerStream + 1, + }); !errors.Is(err, ErrInvalidOutcomeReporter) { + t.Fatalf("NewOutcomeReporter() error = %v, want ErrInvalidOutcomeReporter", err) + } +} + +type outcomeClientStub struct { + batches []*controlplanev1.OutcomeBatch + closeErr error + accepted uint64 +} + +func (client *outcomeClientStub) ReportOutcomes(context.Context) (OutcomeStream, error) { + return &outcomeStreamStub{client: client}, nil +} + +type outcomeStreamStub struct{ client *outcomeClientStub } + +func (stream *outcomeStreamStub) Send(batch *controlplanev1.OutcomeBatch) error { + stream.client.batches = append(stream.client.batches, batch) + return nil +} + +func (stream *outcomeStreamStub) CloseAndRecv() (*controlplanev1.ReportOutcomesResponse, error) { + if stream.client.closeErr != nil { + return nil, stream.client.closeErr + } + accepted := stream.client.accepted + if accepted == 0 && len(stream.client.batches) > 0 { + accepted = stream.client.batches[len(stream.client.batches)-1].GetSequence() + } + return &controlplanev1.ReportOutcomesResponse{AcceptedThroughSequence: accepted}, nil +} + +type retryingOutcomeClient struct { + mu sync.Mutex + batches []*controlplanev1.OutcomeBatch + failures int + confirmed chan struct{} +} + +func (client *retryingOutcomeClient) ReportOutcomes(context.Context) (OutcomeStream, error) { + return &retryingOutcomeStream{client: client}, nil +} + +func (client *retryingOutcomeClient) Batches() []*controlplanev1.OutcomeBatch { + client.mu.Lock() + defer client.mu.Unlock() + return append([]*controlplanev1.OutcomeBatch(nil), client.batches...) +} + +type retryingOutcomeStream struct{ client *retryingOutcomeClient } + +func (stream *retryingOutcomeStream) Send(batch *controlplanev1.OutcomeBatch) error { + stream.client.mu.Lock() + stream.client.batches = append(stream.client.batches, batch) + stream.client.mu.Unlock() + return nil +} + +func (stream *retryingOutcomeStream) CloseAndRecv() (*controlplanev1.ReportOutcomesResponse, error) { + stream.client.mu.Lock() + if stream.client.failures > 0 { + stream.client.failures-- + stream.client.mu.Unlock() + return nil, errors.New("temporary delivery failure") + } + accepted := stream.client.batches[len(stream.client.batches)-1].GetSequence() + confirmed := stream.client.confirmed + stream.client.mu.Unlock() + select { + case confirmed <- struct{}{}: + default: + } + return &controlplanev1.ReportOutcomesResponse{AcceptedThroughSequence: accepted}, nil +} diff --git a/internal/gateway/controlplane/session_runner.go b/internal/gateway/controlplane/session_runner.go index b0b72dd..0b924f3 100644 --- a/internal/gateway/controlplane/session_runner.go +++ b/internal/gateway/controlplane/session_runner.go @@ -10,13 +10,18 @@ var ErrSnapshotStreamClosed = errors.New("worker snapshot stream closed") type SessionRunner struct { reporter *RuntimeReporter watcher *SnapshotWatcher + outcomes *OutcomeReporter } -func NewSessionRunner(reporter *RuntimeReporter, watcher *SnapshotWatcher) (*SessionRunner, error) { - if reporter == nil || watcher == nil { +func NewSessionRunner(reporter *RuntimeReporter, watcher *SnapshotWatcher, outcomes ...*OutcomeReporter) (*SessionRunner, error) { + if reporter == nil || watcher == nil || len(outcomes) > 1 || (len(outcomes) == 1 && outcomes[0] == nil) { return nil, ErrInvalidOptions } - return &SessionRunner{reporter: reporter, watcher: watcher}, nil + runner := &SessionRunner{reporter: reporter, watcher: watcher} + if len(outcomes) == 1 { + runner.outcomes = outcomes[0] + } + return runner, nil } // Run registers exactly once, then runs snapshot intake and Runtime reporting @@ -32,21 +37,36 @@ func (runner *SessionRunner) Run(ctx context.Context) error { } groupCtx, cancel := context.WithCancel(ctx) defer cancel() - results := make(chan error, 2) - go func() { results <- runner.watcher.Watch(groupCtx, registration.SessionID) }() - go func() { results <- runner.reporter.RunRegistered(groupCtx) }() + runners := []func() error{ + func() error { return runner.watcher.Watch(groupCtx, registration.SessionID) }, + func() error { return runner.reporter.RunRegistered(groupCtx) }, + } + if runner.outcomes != nil { + runners = append(runners, func() error { return runner.outcomes.RunRegistered(groupCtx, registration) }) + } + results := make(chan error, len(runners)) + for _, run := range runners { + go func(run func() error) { results <- run() }(run) + } first := <-results cancel() - second := <-results + others := make([]error, 0, len(runners)-1) + for range len(runners) - 1 { + others = append(others, <-results) + } if ctx.Err() != nil { return ctx.Err() } if first == nil { return ErrSnapshotStreamClosed } - if errors.Is(first, context.Canceled) && second != nil { - return second + if errors.Is(first, context.Canceled) { + for _, other := range others { + if other != nil && !errors.Is(other, context.Canceled) { + return other + } + } } return first } diff --git a/internal/gateway/outcome/queue.go b/internal/gateway/outcome/queue.go new file mode 100644 index 0000000..474967e --- /dev/null +++ b/internal/gateway/outcome/queue.go @@ -0,0 +1,99 @@ +// Package outcome provides the Gateway's bounded, non-blocking observation +// buffer. It deliberately has no dependency on Controller or storage code. +package outcome + +import ( + "context" + "errors" + "sync/atomic" + + domain "proxy-pool/internal/domain/outcome" +) + +var ErrInvalidQueueOptions = errors.New("invalid gateway outcome queue options") + +type QueueOptions struct { + Capacity int + MaxBatch int +} + +// Queue accepts observations from the request path without blocking. Events +// beyond Capacity are intentionally dropped because outcome telemetry must not +// delay proxy forwarding or consume unbounded process memory. +type Queue struct { + events chan domain.Event + maxBatch int + dropped atomic.Uint64 +} + +func NewQueue(options QueueOptions) (*Queue, error) { + if options.Capacity <= 0 || options.MaxBatch <= 0 || options.MaxBatch > options.Capacity { + return nil, ErrInvalidQueueOptions + } + return &Queue{events: make(chan domain.Event, options.Capacity), maxBatch: options.MaxBatch}, nil +} + +func (queue *Queue) Record(event domain.Event) { + if queue == nil { + return + } + select { + case queue.events <- event: + default: + queue.dropped.Add(1) + } +} + +func (queue *Queue) Dropped() uint64 { + if queue == nil { + return 0 + } + return queue.dropped.Load() +} + +func (queue *Queue) MaxBatch() int { + if queue == nil { + return 0 + } + return queue.maxBatch +} + +// Next waits for one event then drains an immediately available bounded batch. +func (queue *Queue) Next(ctx context.Context) ([]domain.Event, error) { + if queue == nil || ctx == nil { + return nil, ErrInvalidQueueOptions + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case first := <-queue.events: + return queue.drain(first), nil + } +} + +// TryNext drains one batch only when work is already buffered. +func (queue *Queue) TryNext() ([]domain.Event, bool) { + if queue == nil { + return nil, false + } + select { + case first := <-queue.events: + return queue.drain(first), true + default: + return nil, false + } +} + +func (queue *Queue) drain(first domain.Event) []domain.Event { + batch := make([]domain.Event, 0, queue.maxBatch) + batch = append(batch, first) + for len(batch) < queue.maxBatch { + select { + case event := <-queue.events: + batch = append(batch, event) + default: + return batch + } + } + return batch +} diff --git a/internal/gateway/outcome/queue_test.go b/internal/gateway/outcome/queue_test.go new file mode 100644 index 0000000..0255e82 --- /dev/null +++ b/internal/gateway/outcome/queue_test.go @@ -0,0 +1,41 @@ +package outcome + +import ( + "context" + "errors" + "testing" + + domain "proxy-pool/internal/domain/outcome" +) + +func TestQueueBatchesWithoutBlockingAndCountsOverflow(t *testing.T) { + queue, err := NewQueue(QueueOptions{Capacity: 2, MaxBatch: 2}) + if err != nil { + t.Fatalf("NewQueue() = %v", err) + } + queue.Record(domain.Event{ProxyID: "proxy-a"}) + queue.Record(domain.Event{ProxyID: "proxy-b"}) + queue.Record(domain.Event{ProxyID: "proxy-c"}) + if queue.Dropped() != 1 { + t.Fatalf("Dropped() = %d, want 1", queue.Dropped()) + } + batch, err := queue.Next(context.Background()) + if err != nil || len(batch) != 2 || batch[0].ProxyID != "proxy-a" || batch[1].ProxyID != "proxy-b" { + t.Fatalf("Next() = %+v, %v", batch, err) + } + if batch, ok := queue.TryNext(); ok || batch != nil { + t.Fatalf("TryNext() = %+v, %t; want nil, false", batch, ok) + } +} + +func TestQueueHonorsCanceledContext(t *testing.T) { + queue, err := NewQueue(QueueOptions{Capacity: 1, MaxBatch: 1}) + if err != nil { + t.Fatalf("NewQueue() = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := queue.Next(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("Next(canceled) = %v, want context canceled", err) + } +} diff --git a/internal/gateway/server/handler.go b/internal/gateway/server/handler.go index a2fb329..741002f 100644 --- a/internal/gateway/server/handler.go +++ b/internal/gateway/server/handler.go @@ -14,6 +14,7 @@ import ( "sync/atomic" "time" + outcomeDomain "proxy-pool/internal/domain/outcome" proxyDomain "proxy-pool/internal/domain/proxy" "proxy-pool/internal/domain/routing" "proxy-pool/internal/gateway/dispatch" @@ -78,6 +79,12 @@ type DirectTransport interface { OpenDirectTunnel(context.Context, string) (net.Conn, error) } +// OutcomeRecorder accepts best-effort proxy observations. Implementations must +// return immediately; request forwarding never waits for telemetry delivery. +type OutcomeRecorder interface { + Record(outcomeDomain.Event) +} + type waitingDispatcher interface { AcquireWait(context.Context, dispatch.Request, time.Duration) (*dispatch.Lease, error) } @@ -90,6 +97,7 @@ type Dependencies struct { Router Router Dispatcher Dispatcher Transport ProxyTransport + Outcomes OutcomeRecorder } type Handler struct { @@ -99,6 +107,7 @@ type Handler struct { router Router dispatcher Dispatcher transport ProxyTransport + outcomes OutcomeRecorder buffers sync.Pool inFlight chan struct{} forceClose atomic.Bool @@ -136,6 +145,7 @@ func New(config Config, dependencies Dependencies) (*Handler, error) { router: dependencies.Router, dispatcher: dependencies.Dispatcher, transport: dependencies.Transport, + outcomes: dependencies.Outcomes, tunnels: make(map[*activeTunnel]struct{}), shutdownDone: make(chan struct{}), } @@ -230,14 +240,16 @@ func (handler *Handler) connect( lastErr = directErr break } - handler.serveTunnel(writer, request, nil, upstream) + handler.serveTunnel(writer, request, nil, upstream, "", "") return } break } + started := time.Now().UTC() upstream, err := handler.transport.OpenTunnel(request.Context(), lease.Proxy, target.DialAddress()) if err != nil { + handler.recordOutcome(lease.Proxy.ID, route.RoutingName, outcomeDomain.StageProxyHandshake, false, err, started) finishLease(lease, false) excluded[lease.Proxy.ID] = struct{}{} var responseError *transportDomain.ProxyResponseError @@ -259,7 +271,8 @@ func (handler *Handler) connect( lastErr = err break } - handler.serveTunnel(writer, request, func() { finishLease(lease, true) }, upstream) + handler.recordOutcome(lease.Proxy.ID, route.RoutingName, outcomeDomain.StageProxyHandshake, true, nil, started) + handler.serveTunnel(writer, request, func() { finishLease(lease, true) }, upstream, lease.Proxy.ID, route.RoutingName) return } @@ -279,6 +292,8 @@ func (handler *Handler) serveTunnel( request *http.Request, finish func(), upstream net.Conn, + proxyID string, + routingName string, ) { if finish != nil { defer finish() @@ -287,11 +302,13 @@ func (handler *Handler) serveTunnel( hijacker, ok := writer.(http.Hijacker) if !ok { + handler.recordOutcome(proxyID, routingName, outcomeDomain.StageTunnel, false, errors.New("gateway response does not support connection hijacking"), time.Now().UTC()) http.Error(writer, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError) return } client, readWriter, err := hijacker.Hijack() if err != nil { + handler.recordOutcome(proxyID, routingName, outcomeDomain.StageTunnel, false, err, time.Now().UTC()) http.Error(writer, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError) return } @@ -302,14 +319,18 @@ func (handler *Handler) serveTunnel( } defer handler.unregisterTunnel(tunnel) + started := time.Now().UTC() if _, err := readWriter.WriteString("HTTP/1.1 200 Connection Established\r\n\r\n"); err != nil { + handler.recordOutcome(proxyID, routingName, outcomeDomain.StageTunnel, false, err, started) return } if err := readWriter.Flush(); err != nil { + handler.recordOutcome(proxyID, routingName, outcomeDomain.StageTunnel, false, err, started) return } bufferedClient := &bufferedClientConn{Conn: client, reader: readWriter.Reader} - _ = handler.transport.Relay(request.Context(), bufferedClient, upstream) + err = handler.transport.Relay(request.Context(), bufferedClient, upstream) + handler.recordOutcome(proxyID, routingName, outcomeDomain.StageTunnel, err == nil, err, started) } func (handler *Handler) Shutdown(ctx context.Context) error { @@ -444,8 +465,10 @@ func (handler *Handler) forwardHTTP( committed.Store(true) return nil } + started := time.Now().UTC() response, err := handler.transport.RoundTrip(request.Context(), lease.Proxy, attemptRequest, commit) if err != nil { + handler.recordOutcome(lease.Proxy.ID, route.RoutingName, outcomeDomain.StageDial, false, err, started) finishLease(lease, committed.Load()) excluded[lease.Proxy.ID] = struct{}{} lastErr = err @@ -455,10 +478,17 @@ func (handler *Handler) forwardHTTP( if err := commit(); err != nil { _ = response.Body.Close() finishLease(lease, false) + handler.recordOutcome(lease.Proxy.ID, route.RoutingName, outcomeDomain.StageResponseHeaders, false, err, started) lastErr = err break } } + if response.StatusCode == http.StatusProxyAuthRequired { + handler.recordOutcome(lease.Proxy.ID, route.RoutingName, outcomeDomain.StageResponseHeaders, false, + &transportDomain.ProxyResponseError{StatusCode: response.StatusCode}, started) + } else { + handler.recordOutcome(lease.Proxy.ID, route.RoutingName, outcomeDomain.StageResponseHeaders, true, nil, started) + } handler.writeResponse(writer, response) finishLease(lease, true) return @@ -467,6 +497,61 @@ func (handler *Handler) forwardHTTP( writeGatewayError(writer, fmt.Errorf("forward gateway request: %w", lastErr)) } +func (handler *Handler) recordOutcome( + proxyID, routingName string, + stage outcomeDomain.Stage, + success bool, + err error, + started time.Time, +) { + if handler == nil || handler.outcomes == nil || proxyID == "" { + return + } + observedAt := time.Now().UTC() + if started.IsZero() { + started = observedAt + } + latency := observedAt.Sub(started) + if latency < 0 { + latency = 0 + } + event := outcomeDomain.Event{ + ProxyID: proxyID, RoutingName: routingName, Stage: stage, Success: success, + Latency: latency, ObservedAt: observedAt, + } + if !success { + event.ErrorClass = classifyOutcomeError(stage, err) + } + handler.outcomes.Record(event) +} + +func classifyOutcomeError(stage outcomeDomain.Stage, err error) outcomeDomain.ErrorClass { + switch { + case errors.Is(err, context.Canceled): + return outcomeDomain.ErrorClassCanceled + case errors.Is(err, context.DeadlineExceeded): + return outcomeDomain.ErrorClassTimeout + } + var responseError *transportDomain.ProxyResponseError + if errors.As(err, &responseError) { + return outcomeDomain.ErrorClassProxyResponse + } + var networkError net.Error + if errors.As(err, &networkError) && networkError.Timeout() { + return outcomeDomain.ErrorClassTimeout + } + switch stage { + case outcomeDomain.StageDial: + return outcomeDomain.ErrorClassDial + case outcomeDomain.StageProxyHandshake: + return outcomeDomain.ErrorClassHandshake + case outcomeDomain.StageTunnel: + return outcomeDomain.ErrorClassRelay + default: + return outcomeDomain.ErrorClassInternal + } +} + func (handler *Handler) acquireRoute(ctx context.Context, route dispatch.Request) (*dispatch.Lease, error) { lease, err := handler.dispatcher.Acquire(route) if !errors.Is(err, dispatch.ErrNoCandidate) || route.OnUnavailable != routing.OnUnavailableWait || route.WaitTimeout <= 0 { diff --git a/internal/gateway/server/handler_test.go b/internal/gateway/server/handler_test.go index f575f62..3b6d184 100644 --- a/internal/gateway/server/handler_test.go +++ b/internal/gateway/server/handler_test.go @@ -15,6 +15,7 @@ import ( "testing" "time" + outcomeDomain "proxy-pool/internal/domain/outcome" proxyDomain "proxy-pool/internal/domain/proxy" "proxy-pool/internal/domain/routing" "proxy-pool/internal/gateway/dispatch" @@ -103,6 +104,50 @@ func TestHandlerRetriesGETWithAnotherProxyBeforeResponseCommit(t *testing.T) { assertNoLeakedCapacity(t, view) } +func TestHandlerRecordsOutcomeForEachProxyAttempt(t *testing.T) { + dispatcher, _ := dispatcherWithProxies(t, "proxy-a", "proxy-b") + transport := &fakeTransport{roundTrip: func( + _ context.Context, + selected proxyDomain.Proxy, + _ *http.Request, + commit ...func() error, + ) (*http.Response, error) { + if selected.ID == "proxy-a" { + return nil, errors.New("dial failed") + } + if err := commit[0](); err != nil { + return nil, err + } + return &http.Response{StatusCode: http.StatusNoContent, Header: make(http.Header), Body: http.NoBody}, nil + }} + handler := newTestHandler(t, Config{MaxAttempts: 2, RetryMethods: []string{http.MethodGet}}, dispatcher, transport) + recorder := &outcomeRecorder{} + handler.outcomes = recorder + + response := httptest.NewRecorder() + handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "http://example.test/resource", nil)) + events := recorder.Events() + if response.Code != http.StatusNoContent || len(events) != 2 { + t.Fatalf("status=%d events=%+v", response.Code, events) + } + if events[0].ProxyID != "proxy-a" || events[0].Stage != outcomeDomain.StageDial || events[0].Success || + events[0].ErrorClass != outcomeDomain.ErrorClassDial || events[1].ProxyID != "proxy-b" || + events[1].Stage != outcomeDomain.StageResponseHeaders || !events[1].Success { + t.Fatalf("outcomes = %+v", events) + } +} + +func TestHandlerClampsNegativeOutcomeLatency(t *testing.T) { + recorder := &outcomeRecorder{} + handler := &Handler{outcomes: recorder} + handler.recordOutcome("proxy-a", "route-a", outcomeDomain.StageDial, true, nil, time.Now().Add(time.Second)) + + events := recorder.Events() + if len(events) != 1 || events[0].Latency != 0 { + t.Fatalf("outcomes = %+v, want zero latency", events) + } +} + func TestHandlerDoesNotRetryPOST(t *testing.T) { t.Parallel() @@ -149,6 +194,8 @@ func TestHandlerReturns407WithoutRetry(t *testing.T) { }, nil }} handler := newTestHandler(t, Config{MaxAttempts: 2, RetryMethods: []string{http.MethodGet}}, dispatcher, transport) + recorder := &outcomeRecorder{} + handler.outcomes = recorder response := httptest.NewRecorder() handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "http://example.test/resource", nil)) @@ -159,6 +206,11 @@ func TestHandlerReturns407WithoutRetry(t *testing.T) { if len(transport.attempts()) != 1 { t.Fatalf("attempts = %v, want no retry for 407", transport.attempts()) } + events := recorder.Events() + if len(events) != 1 || events[0].Success || events[0].Stage != outcomeDomain.StageResponseHeaders || + events[0].ErrorClass != outcomeDomain.ErrorClassProxyResponse { + t.Fatalf("outcomes = %+v, want failed proxy response", events) + } assertNoLeakedCapacity(t, view) } @@ -496,6 +548,23 @@ type fakeTransport struct { relay func(context.Context, net.Conn, net.Conn) error } +type outcomeRecorder struct { + mu sync.Mutex + events []outcomeDomain.Event +} + +func (recorder *outcomeRecorder) Record(event outcomeDomain.Event) { + recorder.mu.Lock() + recorder.events = append(recorder.events, event) + recorder.mu.Unlock() +} + +func (recorder *outcomeRecorder) Events() []outcomeDomain.Event { + recorder.mu.Lock() + defer recorder.mu.Unlock() + return append([]outcomeDomain.Event(nil), recorder.events...) +} + type waitRecordingDispatcher struct{ waitTimeout time.Duration } func (dispatcher *waitRecordingDispatcher) Acquire(dispatch.Request) (*dispatch.Lease, error) { @@ -588,13 +657,13 @@ func dispatcherWithProxies(t *testing.T, ids ...string) (*dispatch.Dispatcher, * proxies := make([]proxyDomain.Proxy, 0, len(ids)) for index, id := range ids { proxies = append(proxies, proxyDomain.Proxy{ - ID: id, - Scheme: proxyDomain.SchemeHTTP, - Host: "127.0.0.1", - Port: uint16(20000 + index), - SourceUpstream: "provider-a", - MaxConcurrency: 2, - State: proxyDomain.StateAvailable, + ID: id, + Scheme: proxyDomain.SchemeHTTP, + Host: "127.0.0.1", + Port: uint16(20000 + index), + SourceUpstream: "provider-a", + MaxConcurrency: 2, + State: proxyDomain.StateAvailable, }) } envelope := snapshot.Envelope{