youfak/proxy-pool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: youfak:a79d030c82f97ac7f89f4b88e144608bb0e3c2c4
Branches Tags
youfak:build/proxy-pool-architecture
..
compare: youfak:081e1729706eb18162a94dac24dac04b7d13b9e7
Branches Tags
youfak:build/proxy-pool-architecture

No commits in common. "a79d030c82f97ac7f89f4b88e144608bb0e3c2c4" and "081e1729706eb18162a94dac24dac04b7d13b9e7" have entirely different histories.
a79d030c82 ... 081e172970

23 changed files with 104 additions and 4112 deletions
Show Stats Expand all files Collapse all files

11
.github/workflows/ci.yml vendored
View File

@ -8,17 +8,6 @@ permissions:
contents: read contents: read
jobs: jobs:
proto:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-go@v5
with:
go-version-file: go.mod
cache: true
- shell: pwsh
run: ./scripts/verify-proto.ps1
test: test:
strategy: strategy:
matrix: matrix:

3
configs/proxy-pool.yaml
View File

@ -81,9 +81,6 @@ admin:
auth: auth:
mode: none mode: none
controlPlane:
enabled: false
metrics: metrics:
enabled: true enabled: true
listen: 127.0.0.1:9090 listen: 127.0.0.1:9090

3
deploy/config/local.yaml
View File

@ -58,9 +58,6 @@ admin:
header: X-Admin-Key header: X-Admin-Key
token: "${PROXY_POOL_ADMIN_TOKEN}" token: "${PROXY_POOL_ADMIN_TOKEN}"
controlPlane:
enabled: false
metrics: metrics:
enabled: true enabled: true
listen: 0.0.0.0:9090 listen: 0.0.0.0:9090

2
deploy/kubernetes/base/configmap.yaml
View File

@ -60,8 +60,6 @@ data:
mode: apiKey mode: apiKey
header: X-Admin-Key header: X-Admin-Key
token: "${PROXY_POOL_ADMIN_TOKEN}" token: "${PROXY_POOL_ADMIN_TOKEN}"
controlPlane:
enabled: false
metrics: metrics:
enabled: true enabled: true
listen: 0.0.0.0:9090 listen: 0.0.0.0:9090

76
docs/configuration/reference.md
View File

@ -52,7 +52,6 @@ security: {}
gateway: {} gateway: {}
distribution: {} distribution: {}
admin: {} admin: {}
controlPlane: {}
metrics: {} metrics: {}
storage: {} storage: {}
routing: [] routing: []
@ -65,7 +64,6 @@ upstreams: {}
- `gateway`：HTTP/HTTPS CONNECT 数据面入口。 - `gateway`：HTTP/HTTPS CONNECT 数据面入口。
- `distribution`：一次性独占提取入口。 - `distribution`：一次性独占提取入口。
- `admin`：运维管理入口，必须与 Distribution 分端口。 - `admin`：运维管理入口，必须与 Distribution 分端口。
- `controlPlane`：Worker 注册、Snapshot ACK 和 Runtime 心跳的 gRPC 控制面，默认关闭。
- `metrics`：Prometheus 入口。 - `metrics`：Prometheus 入口。
- `storage`：Controller 使用的 PostgreSQL 与 Redis 地址。 - `storage`：Controller 使用的 PostgreSQL 与 Redis 地址。
- `routing`：有序 Routing 列表，自上而下首条命中停止。 - `routing`：有序 Routing 列表，自上而下首条命中停止。
@ -124,59 +122,7 @@ Lua 可精确表示的整数范围内。
Gateway、Distribution、Admin 与 Provider API 是独立认证边界。改变其中一套 Gateway、Distribution、Admin 与 Provider API 是独立认证边界。改变其中一套
不得连带改变其他入口。 不得连带改变其他入口。
## 4. Worker 控制面 ## 4. Gateway
控制面默认关闭；默认配置中的 `8443` 端口预留不表示服务已监听。控制面 Session、
Snapshot ACK 和 Runtime 报告是 Redis 的短效运行时状态，**不写入 PostgreSQL**。
仅本机开发可以使用回环明文：
```yaml
controlPlane:
enabled: true
listen: 127.0.0.1:8443
protocolVersion: 1
heartbeatInterval: 10s
sessionTTL: 30s
maxStaleAge: 10s
maxMessageBytes: 1048576
maxRuntimeCounters: 100000
maxConcurrentStreams: 128
tls:
mode: disabled
```
任何非回环监听地址都必须使用 mTLS：
```yaml
controlPlane:
enabled: true
listen: 0.0.0.0:8443
protocolVersion: 1
heartbeatInterval: 10s
sessionTTL: 30s
maxStaleAge: 10s
maxMessageBytes: 1048576
maxRuntimeCounters: 100000
maxConcurrentStreams: 128
tls:
mode: mtls
certFile: /run/secrets/controller-cert.pem
keyFile: /run/secrets/controller-key.pem
clientCAFile: /run/secrets/worker-ca.pem
trustDomain: proxy.example
environment: production
```
- `protocolVersion` 当前固定为 `1`。
- `sessionTTL` 至少为 `3 * heartbeatInterval`；`maxStaleAge` 不得短于心跳间隔。
- `maxMessageBytes` 范围为 `1..67108864`；`maxRuntimeCounters` 范围为
`1..1000000`；`maxConcurrentStreams` 必须为正数。
- `tls.mode` 只能为 `disabled` 或 `mtls`。`disabled` 只允许回环监听；`mtls` 必须
同时配置证书、私钥、客户端 CA、全小写 DNS `trustDomain` 和单 URI 路径段
`environment`。
## 5. Gateway
```yaml ```yaml
gateway: gateway:
@ -205,7 +151,7 @@ gateway:
- 保留地址、CGNAT 与云元数据端点始终拒绝，不能通过私网/链路本地开关放行。 - 保留地址、CGNAT 与云元数据端点始终拒绝，不能通过私网/链路本地开关放行。
- `maxConcurrentConnections` 是入口准入上限，不是 Proxy 容量上限。 - `maxConcurrentConnections` 是入口准入上限，不是 Proxy 容量上限。
## 6. Distribution ## 5. Distribution
```yaml ```yaml
distribution: distribution:
@ -249,7 +195,7 @@ Extraction 是固定的一次性独占行为，**没有** `mode`、`leaseDuratio
启用认证。 启用认证。
- `authenticatedClientOrSourceIP`：优先认证主体，无主体时回退来源地址。 - `authenticatedClientOrSourceIP`：优先认证主体，无主体时回退来源地址。
## 7. Routing ## 6. Routing
```yaml ```yaml
routing: routing:
@ -285,7 +231,7 @@ Sequential 的空计数属于 Upstream，当前索引属于 Routing。只有 Pro
成功、模板成功且合法候选为零时才增加空计数。错误不改变空计数；重复候选 成功、模板成功且合法候选为零时才增加空计数。错误不改变空计数；重复候选
会重置空计数但增加独立 duplicate 指标。 会重置空计数但增加独立 duplicate 指标。
## 8. Upstream ## 7. Upstream
```yaml ```yaml
upstreams: upstreams:
@ -335,7 +281,7 @@ upstreams:
urls: [http://connect.rom.miui.com/generate_204] urls: [http://connect.rom.miui.com/generate_204]
``` ```
### 8.1 Provider 与代理认证 ### 7.1 Provider 与代理认证
- `api.auth` 用于系统访问 Provider API。 - `api.auth` 用于系统访问 Provider API。
- `proxyAuth` 用于最终连接被获取的 Proxy。 - `proxyAuth` 用于最终连接被获取的 Proxy。
@ -369,7 +315,7 @@ proxyAuth:
password: "${PROVIDER_PROXY_PASSWORD}" password: "${PROVIDER_PROXY_PASSWORD}"
``` ```
### 8.2 Pool 与累计额度 ### 7.2 Pool 与累计额度
- `pool.maxSize`：当前系统维护且尚未 EXTRACTED 的 Proxy 硬上限，包括 - `pool.maxSize`：当前系统维护且尚未 EXTRACTED 的 Proxy 硬上限，包括
FETCHED、CHECKING、AVAILABLE、SUSPECT、DRAINING 和 pending expected。 FETCHED、CHECKING、AVAILABLE、SUSPECT、DRAINING 和 pending expected。
@ -398,7 +344,7 @@ proxyAuth:
内存 lease 上限按所有配置 Upstream 的 `pool.maxSize * fetch.maxInFlight` 汇总， 内存 lease 上限按所有配置 Upstream 的 `pool.maxSize * fetch.maxInFlight` 汇总，
配置 reload 只提高上限，不预分配对应内存。 配置 reload 只提高上限，不预分配对应内存。
### 8.3 Fetch 限制 ### 7.3 Fetch 限制
- `estimatedIPsPerCall`：冷启动时每次 Provider 调用预计返回的合法 Proxy 数， - `estimatedIPsPerCall`：冷启动时每次 Provider 调用预计返回的合法 Proxy 数，
同时用于 `pool.maxSize` 的 pending 预占；不得从任意 Query 或 Body 字段推断。 同时用于 `pool.maxSize` 的 pending 预占；不得从任意 Query 或 Body 字段推断。
@ -413,7 +359,7 @@ proxyAuth:
大量缺池信号必须合并成 singleflight 或容量为 1 的通知，不能按 Gateway 请求 大量缺池信号必须合并成 singleflight 或容量为 1 的通知，不能按 Gateway 请求
数量线性触发 Provider API。 数量线性触发 Provider API。
### 8.4 Refill 水位 ### 7.4 Refill 水位
- `reconcileInterval`：无事件时重新读取库存的兜底周期，启动时仍立即检查一次。 - `reconcileInterval`：无事件时重新读取库存的兜底周期，启动时仍立即检查一次。
- `minimumAvailableSlots`：可用并发槽位低于该值时进入补池。 - `minimumAvailableSlots`：可用并发槽位低于该值时进入补池。
@ -424,7 +370,7 @@ proxyAuth:
`requestInterval` 只限制外部 API 调用，不能兼任库存复核周期。pending Proxy `requestInterval` 只限制外部 API 调用，不能兼任库存复核周期。pending Proxy
按 `estimatedIPsPerCall * maxConcurrencyPerProxy` 折算槽位，避免并发补池超量。 按 `estimatedIPsPerCall * maxConcurrencyPerProxy` 折算槽位，避免并发补池超量。
### 8.5 生命周期与健康 ### 7.5 生命周期与健康
- 明确绝对过期时间优先于响应 TTL，响应 TTL 优先于配置 `lifecycle.ttl`。 - 明确绝对过期时间优先于响应 TTL，响应 TTL 优先于配置 `lifecycle.ttl`。
- 距离过期不足 `allocationSafetyMargin` 时停止新分配。 - 距离过期不足 `allocationSafetyMargin` 时停止新分配。
@ -432,7 +378,7 @@ proxyAuth:
- 第一次有意义失败进入 SUSPECT；达到 `maxConsecutiveFailures` 后才进入 - 第一次有意义失败进入 SUSPECT；达到 `maxConsecutiveFailures` 后才进入
UNHEALTHY。 UNHEALTHY。
## 9. 存储、Admin 与 Metrics ## 8. 存储、Admin 与 Metrics
```yaml ```yaml
admin: admin:
@ -469,7 +415,7 @@ Metrics 启用时 `listen` 必须是合法 `host:port`。该入口固定提供 `
服务流量门槛，PostgreSQL 故障由 Admin 接口独立报告。Metrics 开关或监听地址 服务流量门槛，PostgreSQL 故障由 Admin 接口独立报告。Metrics 开关或监听地址
变更需要重启 Controller。 变更需要重启 Controller。
## 10. 启动前校验清单 ## 9. 启动前校验清单
1. `version` 必须为 `1`，未知字段拒绝。 1. `version` 必须为 `1`，未知字段拒绝。
2. 所有启用监听器具有合法 `host:port`。 2. 所有启用监听器具有合法 `host:port`。

2409
gen/controlplane/v1/controlplane.pb.go
View File

File diff suppressed because it is too large Load Diff

429
gen/controlplane/v1/controlplane_grpc.pb.go
View File

@ -1,429 +0,0 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.6.2
// - protoc v7.35.0
// source: controlplane/v1/controlplane.proto
package controlplanev1
import (
context "context"
grpc "google.golang.org/grpc"
codes "google.golang.org/grpc/codes"
status "google.golang.org/grpc/status"
emptypb "google.golang.org/protobuf/types/known/emptypb"
)
// This is a compile-time assertion to ensure that this generated file
// is compatible with the grpc package it is being compiled against.
// Requires gRPC-Go v1.64.0 or later.
const _ = grpc.SupportPackageIsVersion9
const (
WorkerControlPlane_RegisterWorker_FullMethodName = "/proxy_pool.controlplane.v1.WorkerControlPlane/RegisterWorker"
WorkerControlPlane_WatchSnapshots_FullMethodName = "/proxy_pool.controlplane.v1.WorkerControlPlane/WatchSnapshots"
WorkerControlPlane_AcknowledgeSnapshot_FullMethodName = "/proxy_pool.controlplane.v1.WorkerControlPlane/AcknowledgeSnapshot"
WorkerControlPlane_ReportOutcomes_FullMethodName = "/proxy_pool.controlplane.v1.WorkerControlPlane/ReportOutcomes"
WorkerControlPlane_ReportRuntime_FullMethodName = "/proxy_pool.controlplane.v1.WorkerControlPlane/ReportRuntime"
)
// WorkerControlPlaneClient is the client API for WorkerControlPlane service.
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
//
// WorkerControlPlane distributes immutable, worker-specific snapshots. The
// gateway hot path does not call this service for individual requests.
type WorkerControlPlaneClient interface {
RegisterWorker(ctx context.Context, in *RegisterWorkerRequest, opts ...grpc.CallOption) (*RegisterWorkerResponse, error)
WatchSnapshots(ctx context.Context, in *WatchSnapshotsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[SnapshotEnvelope], error)
AcknowledgeSnapshot(ctx context.Context, in *AcknowledgeSnapshotRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
ReportOutcomes(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[OutcomeBatch, ReportOutcomesResponse], error)
ReportRuntime(ctx context.Context, in *ReportRuntimeRequest, opts ...grpc.CallOption) (*ReportRuntimeResponse, error)
}
type workerControlPlaneClient struct {
cc grpc.ClientConnInterface
}
func NewWorkerControlPlaneClient(cc grpc.ClientConnInterface) WorkerControlPlaneClient {
return &workerControlPlaneClient{cc}
}
func (c *workerControlPlaneClient) RegisterWorker(ctx context.Context, in *RegisterWorkerRequest, opts ...grpc.CallOption) (*RegisterWorkerResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(RegisterWorkerResponse)
err := c.cc.Invoke(ctx, WorkerControlPlane_RegisterWorker_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *workerControlPlaneClient) WatchSnapshots(ctx context.Context, in *WatchSnapshotsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[SnapshotEnvelope], error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
stream, err := c.cc.NewStream(ctx, &WorkerControlPlane_ServiceDesc.Streams[0], WorkerControlPlane_WatchSnapshots_FullMethodName, cOpts...)
if err != nil {
return nil, err
}
x := &grpc.GenericClientStream[WatchSnapshotsRequest, SnapshotEnvelope]{ClientStream: stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
return x, nil
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type WorkerControlPlane_WatchSnapshotsClient = grpc.ServerStreamingClient[SnapshotEnvelope]
func (c *workerControlPlaneClient) AcknowledgeSnapshot(ctx context.Context, in *AcknowledgeSnapshotRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(emptypb.Empty)
err := c.cc.Invoke(ctx, WorkerControlPlane_AcknowledgeSnapshot_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *workerControlPlaneClient) ReportOutcomes(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[OutcomeBatch, ReportOutcomesResponse], error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
stream, err := c.cc.NewStream(ctx, &WorkerControlPlane_ServiceDesc.Streams[1], WorkerControlPlane_ReportOutcomes_FullMethodName, cOpts...)
if err != nil {
return nil, err
}
x := &grpc.GenericClientStream[OutcomeBatch, ReportOutcomesResponse]{ClientStream: stream}
return x, nil
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type WorkerControlPlane_ReportOutcomesClient = grpc.ClientStreamingClient[OutcomeBatch, ReportOutcomesResponse]
func (c *workerControlPlaneClient) ReportRuntime(ctx context.Context, in *ReportRuntimeRequest, opts ...grpc.CallOption) (*ReportRuntimeResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(ReportRuntimeResponse)
err := c.cc.Invoke(ctx, WorkerControlPlane_ReportRuntime_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
// WorkerControlPlaneServer is the server API for WorkerControlPlane service.
// All implementations must embed UnimplementedWorkerControlPlaneServer
// for forward compatibility.
//
// WorkerControlPlane distributes immutable, worker-specific snapshots. The
// gateway hot path does not call this service for individual requests.
type WorkerControlPlaneServer interface {
RegisterWorker(context.Context, *RegisterWorkerRequest) (*RegisterWorkerResponse, error)
WatchSnapshots(*WatchSnapshotsRequest, grpc.ServerStreamingServer[SnapshotEnvelope]) error
AcknowledgeSnapshot(context.Context, *AcknowledgeSnapshotRequest) (*emptypb.Empty, error)
ReportOutcomes(grpc.ClientStreamingServer[OutcomeBatch, ReportOutcomesResponse]) error
ReportRuntime(context.Context, *ReportRuntimeRequest) (*ReportRuntimeResponse, error)
mustEmbedUnimplementedWorkerControlPlaneServer()
}
// UnimplementedWorkerControlPlaneServer must be embedded to have
// forward compatible implementations.
//
// NOTE: this should be embedded by value instead of pointer to avoid a nil
// pointer dereference when methods are called.
type UnimplementedWorkerControlPlaneServer struct{}
func (UnimplementedWorkerControlPlaneServer) RegisterWorker(context.Context, *RegisterWorkerRequest) (*RegisterWorkerResponse, error) {
return nil, status.Error(codes.Unimplemented, "method RegisterWorker not implemented")
}
func (UnimplementedWorkerControlPlaneServer) WatchSnapshots(*WatchSnapshotsRequest, grpc.ServerStreamingServer[SnapshotEnvelope]) error {
return status.Error(codes.Unimplemented, "method WatchSnapshots not implemented")
}
func (UnimplementedWorkerControlPlaneServer) AcknowledgeSnapshot(context.Context, *AcknowledgeSnapshotRequest) (*emptypb.Empty, error) {
return nil, status.Error(codes.Unimplemented, "method AcknowledgeSnapshot not implemented")
}
func (UnimplementedWorkerControlPlaneServer) ReportOutcomes(grpc.ClientStreamingServer[OutcomeBatch, ReportOutcomesResponse]) error {
return status.Error(codes.Unimplemented, "method ReportOutcomes not implemented")
}
func (UnimplementedWorkerControlPlaneServer) ReportRuntime(context.Context, *ReportRuntimeRequest) (*ReportRuntimeResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ReportRuntime not implemented")
}
func (UnimplementedWorkerControlPlaneServer) mustEmbedUnimplementedWorkerControlPlaneServer() {}
func (UnimplementedWorkerControlPlaneServer) testEmbeddedByValue() {}
// UnsafeWorkerControlPlaneServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to WorkerControlPlaneServer will
// result in compilation errors.
type UnsafeWorkerControlPlaneServer interface {
mustEmbedUnimplementedWorkerControlPlaneServer()
}
func RegisterWorkerControlPlaneServer(s grpc.ServiceRegistrar, srv WorkerControlPlaneServer) {
// If the following call panics, it indicates UnimplementedWorkerControlPlaneServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
t.testEmbeddedByValue()
}
s.RegisterService(&WorkerControlPlane_ServiceDesc, srv)
}
func _WorkerControlPlane_RegisterWorker_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(RegisterWorkerRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(WorkerControlPlaneServer).RegisterWorker(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: WorkerControlPlane_RegisterWorker_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(WorkerControlPlaneServer).RegisterWorker(ctx, req.(*RegisterWorkerRequest))
}
return interceptor(ctx, in, info, handler)
}
func _WorkerControlPlane_WatchSnapshots_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(WatchSnapshotsRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(WorkerControlPlaneServer).WatchSnapshots(m, &grpc.GenericServerStream[WatchSnapshotsRequest, SnapshotEnvelope]{ServerStream: stream})
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type WorkerControlPlane_WatchSnapshotsServer = grpc.ServerStreamingServer[SnapshotEnvelope]
func _WorkerControlPlane_AcknowledgeSnapshot_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(AcknowledgeSnapshotRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(WorkerControlPlaneServer).AcknowledgeSnapshot(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: WorkerControlPlane_AcknowledgeSnapshot_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(WorkerControlPlaneServer).AcknowledgeSnapshot(ctx, req.(*AcknowledgeSnapshotRequest))
}
return interceptor(ctx, in, info, handler)
}
func _WorkerControlPlane_ReportOutcomes_Handler(srv interface{}, stream grpc.ServerStream) error {
return srv.(WorkerControlPlaneServer).ReportOutcomes(&grpc.GenericServerStream[OutcomeBatch, ReportOutcomesResponse]{ServerStream: stream})
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type WorkerControlPlane_ReportOutcomesServer = grpc.ClientStreamingServer[OutcomeBatch, ReportOutcomesResponse]
func _WorkerControlPlane_ReportRuntime_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ReportRuntimeRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(WorkerControlPlaneServer).ReportRuntime(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: WorkerControlPlane_ReportRuntime_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(WorkerControlPlaneServer).ReportRuntime(ctx, req.(*ReportRuntimeRequest))
}
return interceptor(ctx, in, info, handler)
}
// WorkerControlPlane_ServiceDesc is the grpc.ServiceDesc for WorkerControlPlane service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
var WorkerControlPlane_ServiceDesc = grpc.ServiceDesc{
ServiceName: "proxy_pool.controlplane.v1.WorkerControlPlane",
HandlerType: (*WorkerControlPlaneServer)(nil),
Methods: []grpc.MethodDesc{
{
MethodName: "RegisterWorker",
Handler: _WorkerControlPlane_RegisterWorker_Handler,
},
{
MethodName: "AcknowledgeSnapshot",
Handler: _WorkerControlPlane_AcknowledgeSnapshot_Handler,
},
{
MethodName: "ReportRuntime",
Handler: _WorkerControlPlane_ReportRuntime_Handler,
},
},
Streams: []grpc.StreamDesc{
{
StreamName: "WatchSnapshots",
Handler: _WorkerControlPlane_WatchSnapshots_Handler,
ServerStreams: true,
},
{
StreamName: "ReportOutcomes",
Handler: _WorkerControlPlane_ReportOutcomes_Handler,
ClientStreams: true,
},
},
Metadata: "controlplane/v1/controlplane.proto",
}
const (
CheckerControlPlane_StreamCheckTasks_FullMethodName = "/proxy_pool.controlplane.v1.CheckerControlPlane/StreamCheckTasks"
CheckerControlPlane_ReportObservations_FullMethodName = "/proxy_pool.controlplane.v1.CheckerControlPlane/ReportObservations"
)
// CheckerControlPlaneClient is the client API for CheckerControlPlane service.
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
//
// CheckerControlPlane hands bounded check work to independently scalable
// checker processes. Observations are facts; only the Controller reducer may
// change authoritative proxy state.
type CheckerControlPlaneClient interface {
StreamCheckTasks(ctx context.Context, in *StreamCheckTasksRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[CheckTask], error)
ReportObservations(ctx context.Context, in *ObservationBatch, opts ...grpc.CallOption) (*ReportObservationsResponse, error)
}
type checkerControlPlaneClient struct {
cc grpc.ClientConnInterface
}
func NewCheckerControlPlaneClient(cc grpc.ClientConnInterface) CheckerControlPlaneClient {
return &checkerControlPlaneClient{cc}
}
func (c *checkerControlPlaneClient) StreamCheckTasks(ctx context.Context, in *StreamCheckTasksRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[CheckTask], error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
stream, err := c.cc.NewStream(ctx, &CheckerControlPlane_ServiceDesc.Streams[0], CheckerControlPlane_StreamCheckTasks_FullMethodName, cOpts...)
if err != nil {
return nil, err
}
x := &grpc.GenericClientStream[StreamCheckTasksRequest, CheckTask]{ClientStream: stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
return x, nil
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type CheckerControlPlane_StreamCheckTasksClient = grpc.ServerStreamingClient[CheckTask]
func (c *checkerControlPlaneClient) ReportObservations(ctx context.Context, in *ObservationBatch, opts ...grpc.CallOption) (*ReportObservationsResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(ReportObservationsResponse)
err := c.cc.Invoke(ctx, CheckerControlPlane_ReportObservations_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
// CheckerControlPlaneServer is the server API for CheckerControlPlane service.
// All implementations must embed UnimplementedCheckerControlPlaneServer
// for forward compatibility.
//
// CheckerControlPlane hands bounded check work to independently scalable
// checker processes. Observations are facts; only the Controller reducer may
// change authoritative proxy state.
type CheckerControlPlaneServer interface {
StreamCheckTasks(*StreamCheckTasksRequest, grpc.ServerStreamingServer[CheckTask]) error
ReportObservations(context.Context, *ObservationBatch) (*ReportObservationsResponse, error)
mustEmbedUnimplementedCheckerControlPlaneServer()
}
// UnimplementedCheckerControlPlaneServer must be embedded to have
// forward compatible implementations.
//
// NOTE: this should be embedded by value instead of pointer to avoid a nil
// pointer dereference when methods are called.
type UnimplementedCheckerControlPlaneServer struct{}
func (UnimplementedCheckerControlPlaneServer) StreamCheckTasks(*StreamCheckTasksRequest, grpc.ServerStreamingServer[CheckTask]) error {
return status.Error(codes.Unimplemented, "method StreamCheckTasks not implemented")
}
func (UnimplementedCheckerControlPlaneServer) ReportObservations(context.Context, *ObservationBatch) (*ReportObservationsResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ReportObservations not implemented")
}
func (UnimplementedCheckerControlPlaneServer) mustEmbedUnimplementedCheckerControlPlaneServer() {}
func (UnimplementedCheckerControlPlaneServer) testEmbeddedByValue() {}
// UnsafeCheckerControlPlaneServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to CheckerControlPlaneServer will
// result in compilation errors.
type UnsafeCheckerControlPlaneServer interface {
mustEmbedUnimplementedCheckerControlPlaneServer()
}
func RegisterCheckerControlPlaneServer(s grpc.ServiceRegistrar, srv CheckerControlPlaneServer) {
// If the following call panics, it indicates UnimplementedCheckerControlPlaneServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
t.testEmbeddedByValue()
}
s.RegisterService(&CheckerControlPlane_ServiceDesc, srv)
}
func _CheckerControlPlane_StreamCheckTasks_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(StreamCheckTasksRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(CheckerControlPlaneServer).StreamCheckTasks(m, &grpc.GenericServerStream[StreamCheckTasksRequest, CheckTask]{ServerStream: stream})
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type CheckerControlPlane_StreamCheckTasksServer = grpc.ServerStreamingServer[CheckTask]
func _CheckerControlPlane_ReportObservations_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ObservationBatch)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(CheckerControlPlaneServer).ReportObservations(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: CheckerControlPlane_ReportObservations_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(CheckerControlPlaneServer).ReportObservations(ctx, req.(*ObservationBatch))
}
return interceptor(ctx, in, info, handler)
}
// CheckerControlPlane_ServiceDesc is the grpc.ServiceDesc for CheckerControlPlane service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
var CheckerControlPlane_ServiceDesc = grpc.ServiceDesc{
ServiceName: "proxy_pool.controlplane.v1.CheckerControlPlane",
HandlerType: (*CheckerControlPlaneServer)(nil),
Methods: []grpc.MethodDesc{
{
MethodName: "ReportObservations",
Handler: _CheckerControlPlane_ReportObservations_Handler,
},
},
Streams: []grpc.StreamDesc{
{
StreamName: "StreamCheckTasks",
Handler: _CheckerControlPlane_StreamCheckTasks_Handler,
ServerStreams: true,
},
},
Metadata: "controlplane/v1/controlplane.proto",
}

13
go.mod
View File

@ -7,8 +7,6 @@ require (
github.com/prometheus/client_golang v1.24.1 github.com/prometheus/client_golang v1.24.1
github.com/redis/go-redis/v9 v9.19.0 github.com/redis/go-redis/v9 v9.19.0
go.yaml.in/yaml/v4 v4.0.0-rc.3 go.yaml.in/yaml/v4 v4.0.0-rc.3
google.golang.org/grpc v1.83.0
google.golang.org/protobuf v1.36.11
) )
require ( require (
@ -22,16 +20,9 @@ require (
github.com/prometheus/common v0.70.1 // indirect github.com/prometheus/common v0.70.1 // indirect
github.com/prometheus/procfs v0.21.1 // indirect github.com/prometheus/procfs v0.21.1 // indirect
go.uber.org/atomic v1.11.0 // indirect go.uber.org/atomic v1.11.0 // indirect
golang.org/x/crypto v0.54.0 // indirect golang.org/x/crypto v0.17.0 // indirect
golang.org/x/net v0.57.0 // indirect
golang.org/x/sync v0.22.0 // indirect golang.org/x/sync v0.22.0 // indirect
golang.org/x/sys v0.47.0 // indirect golang.org/x/sys v0.47.0 // indirect
golang.org/x/text v0.40.0 // indirect golang.org/x/text v0.40.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/protobuf v1.36.11 // indirect
google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2 // indirect
)
tool (
google.golang.org/grpc/cmd/protoc-gen-go-grpc
google.golang.org/protobuf/cmd/protoc-gen-go
) )

34
go.sum
View File

@ -9,16 +9,8 @@ github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XL
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk= github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
@ -54,18 +46,6 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s=
go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU=
go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc=
go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc=
go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo=
go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58=
go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0=
go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI=
go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA=
go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk=
go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE=
go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
@ -74,24 +54,14 @@ go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
go.yaml.in/yaml/v4 v4.0.0-rc.3 h1:3h1fjsh1CTAPjW7q/EMe+C8shx5d8ctzZTrLcs/j8Go= go.yaml.in/yaml/v4 v4.0.0-rc.3 h1:3h1fjsh1CTAPjW7q/EMe+C8shx5d8ctzZTrLcs/j8Go=
go.yaml.in/yaml/v4 v4.0.0-rc.3/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0= go.yaml.in/yaml/v4 v4.0.0-rc.3/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0=
golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw= golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k=
golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk= golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY= golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
google.golang.org/grpc v1.83.0 h1:JeNZEKJFbQxArAMl+hiytHauacDNqJUllNfmIMmpqnQ=
google.golang.org/grpc v1.83.0/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ=
google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2 h1:rgSNvqscFZ1JgV/4wH5GOsZFSFkR2Eua9As3KIr2LlM=
google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2/go.mod h1:iMEtFwDlAhjDU9L5mY6U1XLwlIId/G3h+QcBHDIvrJ8=
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

23
internal/config/config.go
View File

@ -31,7 +31,6 @@ type Config struct {
Gateway Listener `yaml:"gateway"` Gateway Listener `yaml:"gateway"`
Distribution Distribution `yaml:"distribution"` Distribution Distribution `yaml:"distribution"`
Admin Listener `yaml:"admin"` Admin Listener `yaml:"admin"`
ControlPlane ControlPlane `yaml:"controlPlane"`
Metrics Metrics `yaml:"metrics"` Metrics Metrics `yaml:"metrics"`
Storage Storage `yaml:"storage"` Storage Storage `yaml:"storage"`
Routing []Routing `yaml:"routing"` Routing []Routing `yaml:"routing"`
@ -128,28 +127,6 @@ type Metrics struct {
Listen string `yaml:"listen"` Listen string `yaml:"listen"`
} }
type ControlPlane struct {
Enabled bool `yaml:"enabled"`
Listen string `yaml:"listen"`
ProtocolVersion uint32 `yaml:"protocolVersion"`
HeartbeatInterval Duration `yaml:"heartbeatInterval"`
SessionTTL Duration `yaml:"sessionTTL"`
MaxStaleAge Duration `yaml:"maxStaleAge"`
MaxMessageBytes int `yaml:"maxMessageBytes"`
MaxRuntimeCounters int `yaml:"maxRuntimeCounters"`
MaxConcurrentStreams uint32 `yaml:"maxConcurrentStreams"`
TLS ControlPlaneTLS `yaml:"tls"`
}
type ControlPlaneTLS struct {
Mode string `yaml:"mode"`
CertFile string `yaml:"certFile"`
KeyFile string `yaml:"keyFile"`
ClientCAFile string `yaml:"clientCAFile"`
TrustDomain string `yaml:"trustDomain"`
Environment string `yaml:"environment"`
}
type Storage struct { type Storage struct {
PostgresURL string `yaml:"postgresURL"` PostgresURL string `yaml:"postgresURL"`
RedisURL string `yaml:"redisURL"` RedisURL string `yaml:"redisURL"`

144
internal/config/config_test.go
View File

@ -352,150 +352,6 @@ func TestValidateMetricsListener(t *testing.T) {
} }
} }
func TestValidateControlPlane(t *testing.T) {
tests := []struct {
name string
mutate func(*Config)
want string
}{
{
name: "missing listen",
mutate: func(cfg *Config) {
cfg.ControlPlane = validControlPlane()
cfg.ControlPlane.Listen = ""
},
want: "controlPlane listen",
},
{
name: "public plaintext",
mutate: func(cfg *Config) {
cfg.ControlPlane = validControlPlane()
cfg.ControlPlane.Listen = "0.0.0.0:8443"
},
want: "requires mtls",
},
{
name: "short session ttl",
mutate: func(cfg *Config) {
cfg.ControlPlane = validControlPlane()
cfg.ControlPlane.SessionTTL = Duration(29 * time.Second)
},
want: "sessionTTL",
},
{
name: "stale below heartbeat",
mutate: func(cfg *Config) {
cfg.ControlPlane = validControlPlane()
cfg.ControlPlane.MaxStaleAge = Duration(9 * time.Second)
},
want: "maxStaleAge",
},
{
name: "invalid protocol",
mutate: func(cfg *Config) {
cfg.ControlPlane = validControlPlane()
cfg.ControlPlane.ProtocolVersion = 0
},
want: "protocolVersion",
},
{
name: "invalid message limit",
mutate: func(cfg *Config) {
cfg.ControlPlane = validControlPlane()
cfg.ControlPlane.MaxMessageBytes = 64<<20 + 1
},
want: "maxMessageBytes",
},
{
name: "invalid counter limit",
mutate: func(cfg *Config) {
cfg.ControlPlane = validControlPlane()
cfg.ControlPlane.MaxRuntimeCounters = MaximumPoolSize + 1
},
want: "maxRuntimeCounters",
},
{
name: "zero concurrent streams",
mutate: func(cfg *Config) {
cfg.ControlPlane = validControlPlane()
cfg.ControlPlane.MaxConcurrentStreams = 0
},
want: "maxConcurrentStreams",
},
{
name: "unsupported tls mode",
mutate: func(cfg *Config) {
cfg.ControlPlane = validControlPlane()
cfg.ControlPlane.TLS.Mode = "serverTLS"
},
want: "tls.mode",
},
{
name: "missing mtls files",
mutate: func(cfg *Config) {
cfg.ControlPlane = validMTLSControlPlane()
cfg.ControlPlane.TLS.CertFile = ""
},
want: "certFile",
},
{
name: "trust domain with port",
mutate: func(cfg *Config) {
cfg.ControlPlane = validMTLSControlPlane()
cfg.ControlPlane.TLS.TrustDomain = "proxy.example:443"
},
want: "trustDomain",
},
{
name: "environment path",
mutate: func(cfg *Config) {
cfg.ControlPlane = validMTLSControlPlane()
cfg.ControlPlane.TLS.Environment = "prod/eu"
},
want: "environment",
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
cfg := mustLoadValidConfig(t)
test.mutate(cfg)
if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), test.want) {
t.Fatalf("Validate() error = %v, want substring %q", err, test.want)
}
})
}
}
func validControlPlane() ControlPlane {
return ControlPlane{
Enabled: true,
Listen: "127.0.0.1:8443",
ProtocolVersion: 1,
HeartbeatInterval: Duration(10 * time.Second),
SessionTTL: Duration(30 * time.Second),
MaxStaleAge: Duration(10 * time.Second),
MaxMessageBytes: 1 << 20,
MaxRuntimeCounters: 100_000,
MaxConcurrentStreams: 128,
TLS: ControlPlaneTLS{Mode: "disabled"},
}
}
func validMTLSControlPlane() ControlPlane {
controlPlane := validControlPlane()
controlPlane.Listen = "0.0.0.0:8443"
controlPlane.TLS = ControlPlaneTLS{
Mode: "mtls",
CertFile: "/run/secrets/controller-cert.pem",
KeyFile: "/run/secrets/controller-key.pem",
ClientCAFile: "/run/secrets/worker-ca.pem",
TrustDomain: "proxy.example",
Environment: "production",
}
return controlPlane
}
func TestValidateRejectsInvalidConfigurationMatrix(t *testing.T) { func TestValidateRejectsInvalidConfigurationMatrix(t *testing.T) {
tests := []struct { tests := []struct {
name string name string

72
internal/config/validate.go
View File

@ -10,11 +10,6 @@ import (
"strings" "strings"
) )
var (
controlPlaneIdentityPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$`)
trustDomainLabelPattern = regexp.MustCompile(`^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$`)
)
func Validate(cfg *Config) error { func Validate(cfg *Config) error {
if cfg == nil { if cfg == nil {
return fmt.Errorf("validate configuration: nil config") return fmt.Errorf("validate configuration: nil config")
@ -43,9 +38,6 @@ func Validate(cfg *Config) error {
return err return err
} }
} }
if err := validateControlPlane(cfg.ControlPlane); err != nil {
return err
}
if fetchConfigured(cfg.Defaults.Fetch) { if fetchConfigured(cfg.Defaults.Fetch) {
if err := validateFetch("defaults.fetch", cfg.Defaults.Fetch); err != nil { if err := validateFetch("defaults.fetch", cfg.Defaults.Fetch); err != nil {
return err return err
@ -109,70 +101,6 @@ func Validate(cfg *Config) error {
return nil return nil
} }
func validateControlPlane(item ControlPlane) error {
if !item.Enabled {
return nil
}
host, err := validateListenAddress("controlPlane", item.Listen)
if err != nil {
return err
}
if item.ProtocolVersion != 1 {
return fmt.Errorf("validate controlPlane protocolVersion: must be 1")
}
heartbeat := item.HeartbeatInterval.Value()
if heartbeat <= 0 {
return fmt.Errorf("validate controlPlane heartbeatInterval: must be greater than zero")
}
if item.SessionTTL.Value()/3 < heartbeat {
return fmt.Errorf("validate controlPlane sessionTTL: must be at least three heartbeat intervals")
}
if item.MaxStaleAge.Value() < heartbeat {
return fmt.Errorf("validate controlPlane maxStaleAge: must not be shorter than heartbeatInterval")
}
if item.MaxMessageBytes <= 0 || item.MaxMessageBytes > 64<<20 {
return fmt.Errorf("validate controlPlane maxMessageBytes: must be in [1, 67108864]")
}
if item.MaxRuntimeCounters <= 0 || item.MaxRuntimeCounters > MaximumPoolSize {
return fmt.Errorf("validate controlPlane maxRuntimeCounters: must be in [1, %d]", MaximumPoolSize)
}
if item.MaxConcurrentStreams == 0 {
return fmt.Errorf("validate controlPlane maxConcurrentStreams: must be positive")
}
switch item.TLS.Mode {
case "disabled":
if isPublicHost(host) {
return fmt.Errorf("validate controlPlane tls: non-loopback listen requires mtls")
}
case "mtls":
if item.TLS.CertFile == "" || item.TLS.KeyFile == "" || item.TLS.ClientCAFile == "" ||
item.TLS.TrustDomain == "" || item.TLS.Environment == "" {
return fmt.Errorf("validate controlPlane tls: mtls requires certFile, keyFile, clientCAFile, trustDomain and environment")
}
if !validTrustDomain(item.TLS.TrustDomain) {
return fmt.Errorf("validate controlPlane tls.trustDomain: must be a lowercase DNS name without port")
}
if !controlPlaneIdentityPattern.MatchString(item.TLS.Environment) {
return fmt.Errorf("validate controlPlane tls.environment: must be one URI path segment")
}
default:
return fmt.Errorf("validate controlPlane tls.mode: must be disabled or mtls")
}
return nil
}
func validTrustDomain(value string) bool {
if len(value) == 0 || len(value) > 253 {
return false
}
for _, label := range strings.Split(value, ".") {
if !trustDomainLabelPattern.MatchString(label) {
return false
}
}
return true
}
func validateListener(name string, listener Listener, security Security) error { func validateListener(name string, listener Listener, security Security) error {
if !listener.Enabled { if !listener.Enabled {
return nil return nil

24
internal/domain/workerruntime/contract_external_test.go
View File

@ -1,24 +0,0 @@
package workerruntime_test
import (
"testing"
"time"
"proxy-pool/internal/domain/workerruntime"
"proxy-pool/internal/domain/workerruntime/contracttest"
)
func TestMemoryStoreContract(t *testing.T) {
now := time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC)
contracttest.Run(t, func(t *testing.T) contracttest.Fixture {
t.Helper()
store, err := workerruntime.NewMemoryStore(func() time.Time { return now })
if err != nil {
t.Fatalf("NewMemoryStore(): %v", err)
}
return contracttest.Fixture{
Store: store, Reader: store,
Advance: func(duration time.Duration) { now = now.Add(duration) },
}
})
}

131
internal/domain/workerruntime/contracttest/contract.go
View File

@ -1,131 +0,0 @@
package contracttest
import (
"context"
"crypto/sha256"
"errors"
"testing"
"time"
"proxy-pool/internal/domain/workerruntime"
)
type Fixture struct {
Store workerruntime.ControlStore
Reader workerruntime.RuntimeReader
Advance func(time.Duration)
}
type Factory func(*testing.T) Fixture
// Run exercises the public control-store behavior shared by Memory and Redis.
func Run(t *testing.T, factory Factory) {
t.Helper()
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)) })
}
func runLifecycle(t *testing.T, fixture Fixture) {
t.Helper()
ctx := context.Background()
open(t, fixture.Store)
epoch, err := fixture.Store.CurrentOwnershipEpoch(ctx)
if err != nil {
t.Fatalf("CurrentOwnershipEpoch(): %v", err)
}
reference := snapshot(7, epoch, "snapshot-7")
report := runtimeReport(1, 7, epoch)
if err := fixture.Store.ReplaceRuntime(ctx, report, time.Minute); !errors.Is(err, workerruntime.ErrSnapshotMismatch) {
t.Fatalf("ReplaceRuntime(before ACK) error = %v", err)
}
if err := fixture.Store.RecordIssuedSnapshot(ctx, reference, time.Minute); err != nil {
t.Fatalf("RecordIssuedSnapshot(): %v", err)
}
ack := workerruntime.SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: reference, Applied: true}
if err := fixture.Store.AcknowledgeSnapshot(ctx, ack, time.Minute); err != nil {
t.Fatalf("AcknowledgeSnapshot(): %v", err)
}
if err := fixture.Store.ReplaceRuntime(ctx, report, time.Minute); err != nil {
t.Fatalf("ReplaceRuntime(after ACK): %v", err)
}
assertFresh(t, fixture.Reader, epoch, true)
if err := fixture.Store.AcknowledgeSnapshot(ctx, ack, time.Minute); err != nil {
t.Fatalf("AcknowledgeSnapshot(replay): %v", err)
}
if err := fixture.Store.ReplaceRuntime(ctx, runtimeReport(0, 7, epoch), time.Minute); !errors.Is(err, workerruntime.ErrInvalidReport) {
t.Fatalf("ReplaceRuntime(invalid sequence) error = %v", err)
}
fixture.Advance(2 * time.Minute)
assertFresh(t, fixture.Reader, epoch, false)
}
func runNegativeAck(t *testing.T, fixture Fixture) {
t.Helper()
ctx := context.Background()
open(t, fixture.Store)
epoch, err := fixture.Store.CurrentOwnershipEpoch(ctx)
if err != nil {
t.Fatalf("CurrentOwnershipEpoch(): %v", err)
}
first := snapshot(7, epoch, "snapshot-7")
if err := fixture.Store.RecordIssuedSnapshot(ctx, first, time.Minute); err != nil {
t.Fatalf("RecordIssuedSnapshot(first): %v", err)
}
if err := fixture.Store.AcknowledgeSnapshot(ctx, workerruntime.SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: first, Applied: true}, time.Minute); err != nil {
t.Fatalf("AcknowledgeSnapshot(first): %v", err)
}
second := snapshot(8, epoch, "snapshot-8")
if err := fixture.Store.RecordIssuedSnapshot(ctx, second, time.Minute); err != nil {
t.Fatalf("RecordIssuedSnapshot(second): %v", err)
}
if err := fixture.Store.AcknowledgeSnapshot(ctx, workerruntime.SnapshotAcknowledgement{
WorkerID: "worker-a", SessionID: "session-a", Reference: second, ErrorCode: "apply_failed",
}, time.Minute); err != nil {
t.Fatalf("AcknowledgeSnapshot(negative): %v", err)
}
if err := fixture.Store.ReplaceRuntime(ctx, runtimeReport(1, 7, epoch), time.Minute); !errors.Is(err, workerruntime.ErrSnapshotMismatch) {
t.Fatalf("ReplaceRuntime(delayed): %v", err)
}
}
func newFixture(t *testing.T, factory Factory) Fixture {
t.Helper()
fixture := factory(t)
if fixture.Store == nil || fixture.Reader == nil || fixture.Advance == nil {
t.Fatal("contract fixture is incomplete")
}
return fixture
}
func open(t *testing.T, store workerruntime.ControlStore) {
t.Helper()
err := store.OpenSession(context.Background(), workerruntime.Session{
WorkerID: "worker-a", InstanceID: "instance-a", SessionID: "session-a", Zone: "zone-a", ProtocolVersion: 1,
}, time.Minute)
if err != nil {
t.Fatalf("OpenSession(): %v", err)
}
}
func snapshot(version, epoch uint64, value string) workerruntime.SnapshotReference {
return workerruntime.SnapshotReference{
WorkerID: "worker-a", Version: version, OwnershipEpoch: epoch, Checksum: sha256.Sum256([]byte(value)),
}
}
func runtimeReport(sequence, version, epoch uint64) workerruntime.Report {
return workerruntime.Report{
WorkerID: "worker-a", SessionID: "session-a", Sequence: sequence, SnapshotVersion: version,
OwnershipEpoch: epoch, ObservedAt: time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC),
}
}
func assertFresh(t *testing.T, reader workerruntime.RuntimeReader, epoch uint64, want bool) {
t.Helper()
snapshots, err := reader.ReadRuntime(context.Background(), []workerruntime.OwnedProxy{{
ProxyID: "proxy-a", WorkerID: "worker-a", OwnershipEpoch: epoch,
}})
if err != nil || len(snapshots) != 1 || snapshots[0].Fresh != want {
t.Fatalf("ReadRuntime() = %+v, %v; want Fresh=%t", snapshots, err, want)
}
}

286
internal/domain/workerruntime/memory.go
View File

@ -3,17 +3,18 @@ package workerruntime
import ( import (
"context" "context"
"crypto/sha256" "crypto/sha256"
"encoding/json"
"sort"
"strings"
"sync" "sync"
"time" "time"
) )
type MemoryStore struct { type MemoryStore struct {
mu sync.Mutex mu sync.Mutex
now func() time.Time now func() time.Time
epoch uint64 sessions map[string]memorySession
sessions map[string]memorySession reports map[string]memoryReport
references map[string]memoryReference
reports map[string]memoryReport
} }
type memorySession struct { type memorySession struct {
@ -21,11 +22,6 @@ type memorySession struct {
expiresAt time.Time expiresAt time.Time
} }
type memoryReference struct {
value SnapshotReference
expiresAt time.Time
}
type memoryReport struct { type memoryReport struct {
value Report value Report
digest [sha256.Size]byte digest [sha256.Size]byte
@ -34,7 +30,6 @@ type memoryReport struct {
} }
var ( var (
_ ControlStore = (*MemoryStore)(nil)
_ SessionWriter = (*MemoryStore)(nil) _ SessionWriter = (*MemoryStore)(nil)
_ ReportWriter = (*MemoryStore)(nil) _ ReportWriter = (*MemoryStore)(nil)
_ RuntimeReader = (*MemoryStore)(nil) _ RuntimeReader = (*MemoryStore)(nil)
@ -45,184 +40,33 @@ func NewMemoryStore(now func() time.Time) (*MemoryStore, error) {
return nil, ErrInvalidStore return nil, ErrInvalidStore
} }
return &MemoryStore{ return &MemoryStore{
now: now, epoch: 1, sessions: make(map[string]memorySession), now: now, sessions: make(map[string]memorySession), reports: make(map[string]memoryReport),
references: make(map[string]memoryReference), reports: make(map[string]memoryReport),
}, nil }, nil
} }
func (store *MemoryStore) CurrentOwnershipEpoch(ctx context.Context) (uint64, error) {
if ctx == nil || store == nil {
return 0, ErrInvalidStore
}
if err := ctx.Err(); err != nil {
return 0, err
}
store.mu.Lock()
defer store.mu.Unlock()
if store.epoch == 0 {
return 0, ErrInvalidStore
}
return store.epoch, nil
}
// OpenSession always replaces the previous Worker session and clears Runtime.
func (store *MemoryStore) OpenSession(ctx context.Context, session Session, ttl time.Duration) error {
if ctx == nil || store == nil || ttl <= 0 {
return ErrInvalidSession
}
if err := ctx.Err(); err != nil {
return err
}
normalized, err := NormalizeSession(session)
if err != nil {
return err
}
now, err := store.currentTime()
if err != nil {
return err
}
store.mu.Lock()
defer store.mu.Unlock()
delete(store.reports, normalized.WorkerID)
store.sessions[normalized.WorkerID] = memorySession{value: normalized, expiresAt: now.Add(ttl)}
return nil
}
func (store *MemoryStore) RecordIssuedSnapshot(ctx context.Context, reference SnapshotReference, ttl time.Duration) error {
if ctx == nil || store == nil || ttl <= 0 {
return ErrInvalidSnapshotReference
}
if err := ctx.Err(); err != nil {
return err
}
normalized, err := NormalizeSnapshotReference(reference)
if err != nil {
return err
}
now, err := store.currentTime()
if err != nil {
return err
}
store.mu.Lock()
defer store.mu.Unlock()
if normalized.OwnershipEpoch != store.epoch {
return ErrSnapshotMismatch
}
current, exists := store.references[normalized.WorkerID]
if exists && !current.expiresAt.After(now) {
delete(store.references, normalized.WorkerID)
exists = false
}
if exists {
switch compareSnapshotTuple(normalized, current.value) {
case -1:
return ErrStaleSnapshotReference
case 0:
if normalized.Checksum != current.value.Checksum {
return ErrConflictingSnapshotReference
}
}
}
store.references[normalized.WorkerID] = memoryReference{value: normalized, expiresAt: now.Add(ttl)}
return nil
}
func (store *MemoryStore) AcknowledgeSnapshot(ctx context.Context, acknowledgement SnapshotAcknowledgement, ttl time.Duration) error {
if ctx == nil || store == nil || ttl <= 0 {
return ErrInvalidAcknowledgement
}
if err := ctx.Err(); err != nil {
return err
}
normalized, err := NormalizeAcknowledgement(acknowledgement)
if err != nil {
return err
}
now, err := store.currentTime()
if err != nil {
return 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 ErrStaleSession
}
if session.value.AckedSnapshotVersion != 0 {
acknowledged := sessionReference(session.value)
switch compareSnapshotTuple(normalized.Reference, acknowledged) {
case -1:
return ErrStaleAcknowledgement
case 0:
if normalized.Reference.Checksum != acknowledged.Checksum {
return ErrSnapshotMismatch
}
}
}
reference, exists := store.references[normalized.WorkerID]
if !exists || !reference.expiresAt.After(now) {
return ErrSnapshotMismatch
}
switch compareSnapshotTuple(normalized.Reference, reference.value) {
case -1:
return ErrStaleAcknowledgement
case 1:
return ErrSnapshotMismatch
}
if normalized.Reference.Checksum != reference.value.Checksum {
return ErrSnapshotMismatch
}
if !normalized.Applied {
delete(store.reports, normalized.WorkerID)
session.value.RuntimeEnabled = false
session.expiresAt = now.Add(ttl)
store.sessions[normalized.WorkerID] = session
return nil
}
if session.value.AckedSnapshotVersion != 0 && compareSnapshotTuple(normalized.Reference, sessionReference(session.value)) == 0 {
if !session.value.RuntimeEnabled {
delete(store.reports, normalized.WorkerID)
session.value.RuntimeEnabled = true
}
session.expiresAt = now.Add(ttl)
store.sessions[normalized.WorkerID] = session
return nil
}
delete(store.reports, normalized.WorkerID)
session.value.AckedSnapshotVersion = normalized.Reference.Version
session.value.AckedOwnershipEpoch = normalized.Reference.OwnershipEpoch
session.value.AckedChecksum = normalized.Reference.Checksum
session.value.RuntimeEnabled = true
session.expiresAt = now.Add(ttl)
store.sessions[normalized.WorkerID] = session
return nil
}
// ReplaceSession is retained temporarily for the pre-control-plane adapters.
func (store *MemoryStore) ReplaceSession(ctx context.Context, session Session, ttl time.Duration) error { func (store *MemoryStore) ReplaceSession(ctx context.Context, session Session, ttl time.Duration) error {
if ctx == nil || store == nil || !validLegacySession(session) || ttl <= 0 { if ctx == nil || store == nil || !validSession(session) || ttl <= 0 {
return ErrInvalidSession return ErrInvalidSession
} }
if err := ctx.Err(); err != nil { if err := ctx.Err(); err != nil {
return err return err
} }
now, err := store.currentTime() now := store.now().UTC()
if err != nil { if now.IsZero() {
return err return ErrInvalidStore
} }
store.mu.Lock() store.mu.Lock()
defer store.mu.Unlock() defer store.mu.Unlock()
current, exists := store.sessions[session.WorkerID] current, exists := store.sessions[session.WorkerID]
identityChanged := exists && (current.value.SessionID != session.SessionID || current.value.InstanceID != session.InstanceID) identityChanged := exists && (current.value.SessionID != session.SessionID || current.value.InstanceID != session.InstanceID)
expired := exists && !current.expiresAt.After(now) expired := exists && !current.expiresAt.After(now)
if exists && !identityChanged && !expired && legacySessionBefore(session, current.value) { if exists && !identityChanged && !expired && sessionBefore(session, current.value) {
return ErrStaleSession return ErrStaleSession
} }
ackAdvanced := exists && !identityChanged && !expired && legacySessionAfter(session, current.value) ackAdvanced := exists && !identityChanged && !expired && sessionAfter(session, current.value)
if identityChanged || expired || ackAdvanced { if identityChanged || expired || ackAdvanced {
delete(store.reports, session.WorkerID) delete(store.reports, session.WorkerID)
} }
session.RuntimeEnabled = true
store.sessions[session.WorkerID] = memorySession{value: session, expiresAt: now.Add(ttl)} store.sessions[session.WorkerID] = memorySession{value: session, expiresAt: now.Add(ttl)}
return nil return nil
} }
@ -234,48 +78,44 @@ func (store *MemoryStore) ReplaceRuntime(ctx context.Context, report Report, ttl
if err := ctx.Err(); err != nil { if err := ctx.Err(); err != nil {
return err return err
} }
normalized, digest, err := NormalizeReport(report) normalized, counterIndex, err := normalizeReport(report)
if err != nil { if err != nil {
return err return err
} }
now, err := store.currentTime() payload, err := json.Marshal(normalized)
if err != nil { if err != nil {
return err return ErrInvalidReport
}
digest := sha256.Sum256(payload)
now := store.now().UTC()
if now.IsZero() {
return ErrInvalidStore
} }
store.mu.Lock() store.mu.Lock()
defer store.mu.Unlock() defer store.mu.Unlock()
session, exists := store.sessions[normalized.WorkerID] session, exists := store.sessions[report.WorkerID]
if !exists || !session.expiresAt.After(now) || session.value.SessionID != normalized.SessionID { if !exists || !session.expiresAt.After(now) || session.value.SessionID != report.SessionID {
return ErrStaleSession return ErrStaleSession
} }
if !session.value.RuntimeEnabled || normalized.SnapshotVersion != session.value.AckedSnapshotVersion || if report.SnapshotVersion != session.value.AckedSnapshotVersion ||
normalized.OwnershipEpoch != session.value.AckedOwnershipEpoch { report.OwnershipEpoch != session.value.AckedOwnershipEpoch {
return ErrSnapshotMismatch return ErrStaleReport
} }
current, exists := store.reports[normalized.WorkerID] if current, exists := store.reports[report.WorkerID]; exists && current.value.SessionID == report.SessionID {
if exists && !current.expiresAt.After(now) {
delete(store.reports, normalized.WorkerID)
exists = false
}
if exists && current.value.SessionID == normalized.SessionID {
switch { switch {
case normalized.Sequence < current.value.Sequence: case normalized.Sequence < current.value.Sequence:
return ErrStaleReport return ErrStaleReport
case normalized.Sequence == current.value.Sequence && digest != current.digest: case normalized.Sequence == current.value.Sequence && digest != current.digest:
return ErrConflictingReport return ErrConflictingReport
case normalized.Sequence == current.value.Sequence: case normalized.Sequence == current.value.Sequence:
current.expiresAt = now.Add(ttl)
store.reports[normalized.WorkerID] = current
session.expiresAt = now.Add(ttl)
store.sessions[normalized.WorkerID] = session
return nil return nil
} }
} }
store.reports[normalized.WorkerID] = memoryReport{ store.reports[report.WorkerID] = memoryReport{
value: normalized, digest: digest, expiresAt: now.Add(ttl), counters: countersByProxy(normalized.Counters), value: normalized, digest: digest, expiresAt: now.Add(ttl), counters: counterIndex,
} }
session.expiresAt = now.Add(ttl) session.expiresAt = now.Add(ttl)
store.sessions[normalized.WorkerID] = session store.sessions[report.WorkerID] = session
return nil return nil
} }
@ -288,7 +128,7 @@ func (store *MemoryStore) ReadRuntime(ctx context.Context, proxies []OwnedProxy)
} }
seen := make(map[string]struct{}, len(proxies)) seen := make(map[string]struct{}, len(proxies))
for _, proxy := range proxies { for _, proxy := range proxies {
if !ValidIdentifier(proxy.ProxyID) || !ValidIdentifier(proxy.WorkerID) || proxy.OwnershipEpoch == 0 { if !clean(proxy.ProxyID) || !clean(proxy.WorkerID) || proxy.OwnershipEpoch == 0 {
return nil, ErrInvalidQuery return nil, ErrInvalidQuery
} }
key := proxy.WorkerID + "\x00" + proxy.ProxyID key := proxy.WorkerID + "\x00" + proxy.ProxyID
@ -297,9 +137,9 @@ func (store *MemoryStore) ReadRuntime(ctx context.Context, proxies []OwnedProxy)
} }
seen[key] = struct{}{} seen[key] = struct{}{}
} }
now, err := store.currentTime() now := store.now().UTC()
if err != nil { if now.IsZero() {
return nil, err return nil, ErrInvalidStore
} }
store.mu.Lock() store.mu.Lock()
defer store.mu.Unlock() defer store.mu.Unlock()
@ -309,7 +149,7 @@ func (store *MemoryStore) ReadRuntime(ctx context.Context, proxies []OwnedProxy)
session, sessionExists := store.sessions[proxy.WorkerID] session, sessionExists := store.sessions[proxy.WorkerID]
report, reportExists := store.reports[proxy.WorkerID] report, reportExists := store.reports[proxy.WorkerID]
if !sessionExists || !reportExists || !session.expiresAt.After(now) || !report.expiresAt.After(now) || if !sessionExists || !reportExists || !session.expiresAt.After(now) || !report.expiresAt.After(now) ||
!session.value.RuntimeEnabled || report.value.SessionID != session.value.SessionID || report.value.SessionID != session.value.SessionID ||
report.value.SnapshotVersion != session.value.AckedSnapshotVersion || report.value.SnapshotVersion != session.value.AckedSnapshotVersion ||
report.value.OwnershipEpoch != session.value.AckedOwnershipEpoch || report.value.OwnershipEpoch != session.value.AckedOwnershipEpoch ||
report.value.OwnershipEpoch < proxy.OwnershipEpoch { report.value.OwnershipEpoch < proxy.OwnershipEpoch {
@ -325,31 +165,47 @@ func (store *MemoryStore) ReadRuntime(ctx context.Context, proxies []OwnedProxy)
return result, nil return result, nil
} }
func (store *MemoryStore) currentTime() (time.Time, error) { func normalizeReport(report Report) (Report, map[string]Counter, error) {
now := store.now().UTC() if !clean(report.WorkerID) || !clean(report.SessionID) || report.Sequence == 0 ||
if now.IsZero() { report.SnapshotVersion == 0 || report.OwnershipEpoch == 0 || report.ObservedAt.IsZero() {
return time.Time{}, ErrInvalidStore return Report{}, nil, ErrInvalidReport
} }
return now, nil normalized := report
normalized.ObservedAt = report.ObservedAt.UTC()
normalized.Counters = append([]Counter(nil), report.Counters...)
sort.Slice(normalized.Counters, func(left, right int) bool {
return normalized.Counters[left].ProxyID < normalized.Counters[right].ProxyID
})
index := make(map[string]Counter, len(normalized.Counters))
for _, counter := range normalized.Counters {
if !clean(counter.ProxyID) || counter.Active < 0 || counter.Reserved < 0 {
return Report{}, nil, ErrInvalidReport
}
if _, exists := index[counter.ProxyID]; exists {
return Report{}, nil, ErrInvalidReport
}
index[counter.ProxyID] = counter
}
return normalized, index, nil
} }
func countersByProxy(counters []Counter) map[string]Counter { func validSession(session Session) bool {
indexed := make(map[string]Counter, len(counters)) return clean(session.WorkerID) && clean(session.InstanceID) && clean(session.SessionID) &&
for _, counter := range counters {
indexed[counter.ProxyID] = counter
}
return indexed
}
func validLegacySession(session Session) bool {
return ValidIdentifier(session.WorkerID) && ValidIdentifier(session.InstanceID) && ValidIdentifier(session.SessionID) &&
session.AckedSnapshotVersion > 0 && session.AckedOwnershipEpoch > 0 session.AckedSnapshotVersion > 0 && session.AckedOwnershipEpoch > 0
} }
func legacySessionBefore(left, right Session) bool { func sessionBefore(left, right Session) bool {
return compareSnapshotTuple(sessionReference(left), sessionReference(right)) < 0 return left.AckedOwnershipEpoch < right.AckedOwnershipEpoch ||
(left.AckedOwnershipEpoch == right.AckedOwnershipEpoch &&
left.AckedSnapshotVersion < right.AckedSnapshotVersion)
} }
func legacySessionAfter(left, right Session) bool { func sessionAfter(left, right Session) bool {
return compareSnapshotTuple(sessionReference(left), sessionReference(right)) > 0 return left.AckedOwnershipEpoch > right.AckedOwnershipEpoch ||
(left.AckedOwnershipEpoch == right.AckedOwnershipEpoch &&
left.AckedSnapshotVersion > right.AckedSnapshotVersion)
}
func clean(value string) bool {
return value != "" && strings.TrimSpace(value) == value
} }

145
internal/domain/workerruntime/memory_test.go
View File

@ -2,125 +2,11 @@ package workerruntime
import ( import (
"context" "context"
"crypto/sha256"
"errors" "errors"
"testing" "testing"
"time" "time"
) )
func TestMemoryStoreRequiresAcknowledgedSnapshotForRuntime(t *testing.T) {
now := time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC)
store := newRuntimeStore(t, &now)
ctx := context.Background()
openControlSession(t, store, "session-a", time.Minute)
epoch, err := store.CurrentOwnershipEpoch(ctx)
if err != nil {
t.Fatalf("CurrentOwnershipEpoch(): %v", err)
}
report := controlReport(now, "session-a", 1, 7, epoch)
if err := store.ReplaceRuntime(ctx, report, time.Minute); !errors.Is(err, ErrSnapshotMismatch) {
t.Fatalf("ReplaceRuntime(before ACK) error = %v, want ErrSnapshotMismatch", err)
}
reference := controlReference(7, epoch, "snapshot-7")
if err := store.RecordIssuedSnapshot(ctx, reference, time.Minute); err != nil {
t.Fatalf("RecordIssuedSnapshot(): %v", err)
}
if err := store.AcknowledgeSnapshot(ctx, SnapshotAcknowledgement{
WorkerID: "worker-a", SessionID: "session-a", Reference: reference, Applied: true,
}, time.Minute); err != nil {
t.Fatalf("AcknowledgeSnapshot(): %v", err)
}
if err := store.ReplaceRuntime(ctx, report, time.Minute); err != nil {
t.Fatalf("ReplaceRuntime(after ACK): %v", err)
}
}
func TestMemoryStoreNegativeAcknowledgementFencesDelayedRuntime(t *testing.T) {
now := time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC)
store := newRuntimeStore(t, &now)
ctx := context.Background()
openControlSession(t, store, "session-a", time.Minute)
epoch, err := store.CurrentOwnershipEpoch(ctx)
if err != nil {
t.Fatalf("CurrentOwnershipEpoch(): %v", err)
}
first := controlReference(7, epoch, "snapshot-7")
if err := store.RecordIssuedSnapshot(ctx, first, time.Minute); err != nil {
t.Fatalf("RecordIssuedSnapshot(first): %v", err)
}
if err := store.AcknowledgeSnapshot(ctx, SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: first, Applied: true}, time.Minute); err != nil {
t.Fatalf("AcknowledgeSnapshot(first): %v", err)
}
if err := store.ReplaceRuntime(ctx, controlReport(now, "session-a", 1, 7, epoch), time.Minute); err != nil {
t.Fatalf("ReplaceRuntime(first): %v", err)
}
second := controlReference(8, epoch, "snapshot-8")
if err := store.RecordIssuedSnapshot(ctx, second, time.Minute); err != nil {
t.Fatalf("RecordIssuedSnapshot(second): %v", err)
}
if err := store.AcknowledgeSnapshot(ctx, SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: second, Applied: false, ErrorCode: "apply_failed"}, time.Minute); err != nil {
t.Fatalf("AcknowledgeSnapshot(negative): %v", err)
}
if err := store.ReplaceRuntime(ctx, controlReport(now, "session-a", 2, 7, epoch), time.Minute); !errors.Is(err, ErrSnapshotMismatch) {
t.Fatalf("ReplaceRuntime(delayed): %v, want ErrSnapshotMismatch", err)
}
if err := store.AcknowledgeSnapshot(ctx, SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: second, Applied: true}, time.Minute); err != nil {
t.Fatalf("AcknowledgeSnapshot(recover): %v", err)
}
if err := store.ReplaceRuntime(ctx, controlReport(now, "session-a", 2, 8, epoch), time.Minute); err != nil {
t.Fatalf("ReplaceRuntime(recovered): %v", err)
}
}
func TestMemoryStoreAcknowledgementReplayPreservesRuntimeFence(t *testing.T) {
now := time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC)
store := newRuntimeStore(t, &now)
ctx := context.Background()
openControlSession(t, store, "session-a", time.Minute)
epoch, err := store.CurrentOwnershipEpoch(ctx)
if err != nil {
t.Fatalf("CurrentOwnershipEpoch(): %v", err)
}
reference := controlReference(7, epoch, "snapshot-7")
if err := store.RecordIssuedSnapshot(ctx, reference, time.Minute); err != nil {
t.Fatalf("RecordIssuedSnapshot(): %v", err)
}
ack := SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: reference, Applied: true}
if err := store.AcknowledgeSnapshot(ctx, ack, time.Minute); err != nil {
t.Fatalf("AcknowledgeSnapshot(first): %v", err)
}
if err := store.ReplaceRuntime(ctx, controlReport(now, "session-a", 100, 7, epoch), time.Minute); err != nil {
t.Fatalf("ReplaceRuntime(): %v", err)
}
if err := store.AcknowledgeSnapshot(ctx, ack, time.Minute); err != nil {
t.Fatalf("AcknowledgeSnapshot(replay): %v", err)
}
got, err := store.ReadRuntime(ctx, []OwnedProxy{{ProxyID: "proxy-a", WorkerID: "worker-a", OwnershipEpoch: epoch}})
if err != nil || len(got) != 1 || !got[0].Fresh {
t.Fatalf("ReadRuntime(after ACK replay) = %+v, %v", got, err)
}
if err := store.ReplaceRuntime(ctx, controlReport(now, "session-a", 99, 7, epoch), time.Minute); !errors.Is(err, ErrStaleReport) {
t.Fatalf("ReplaceRuntime(stale): %v, want ErrStaleReport", err)
}
}
func TestMemoryStoreRejectsConflictingSnapshotReference(t *testing.T) {
now := time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC)
store := newRuntimeStore(t, &now)
ctx := context.Background()
epoch, err := store.CurrentOwnershipEpoch(ctx)
if err != nil {
t.Fatalf("CurrentOwnershipEpoch(): %v", err)
}
if err := store.RecordIssuedSnapshot(ctx, controlReference(7, epoch, "first"), time.Minute); err != nil {
t.Fatalf("RecordIssuedSnapshot(first): %v", err)
}
if err := store.RecordIssuedSnapshot(ctx, controlReference(7, epoch, "second"), time.Minute); !errors.Is(err, ErrConflictingSnapshotReference) {
t.Fatalf("RecordIssuedSnapshot(conflict): %v, want ErrConflictingSnapshotReference", err)
}
}
func TestMemoryStoreReplacesSparseRuntimeAndClearsMissingCounters(t *testing.T) { func TestMemoryStoreReplacesSparseRuntimeAndClearsMissingCounters(t *testing.T) {
now := time.Date(2026, 7, 30, 15, 0, 0, 0, time.UTC) now := time.Date(2026, 7, 30, 15, 0, 0, 0, time.UTC)
store := newRuntimeStore(t, &now) store := newRuntimeStore(t, &now)
@ -217,13 +103,13 @@ func TestMemoryStoreRejectsRuntimeBeyondAcknowledgedSnapshot(t *testing.T) {
WorkerID: "worker-a", SessionID: "session-a", Sequence: 1, WorkerID: "worker-a", SessionID: "session-a", Sequence: 1,
SnapshotVersion: 4, OwnershipEpoch: 9, ObservedAt: now, SnapshotVersion: 4, OwnershipEpoch: 9, ObservedAt: now,
} }
if err := store.ReplaceRuntime(context.Background(), report, time.Minute); !errors.Is(err, ErrSnapshotMismatch) { if err := store.ReplaceRuntime(context.Background(), report, time.Minute); !errors.Is(err, ErrStaleReport) {
t.Fatalf("ReplaceRuntime(ahead snapshot) error = %v, want ErrSnapshotMismatch", err) t.Fatalf("ReplaceRuntime(ahead snapshot) error = %v, want ErrStaleReport", err)
} }
report.SnapshotVersion = 3 report.SnapshotVersion = 3
report.OwnershipEpoch = 10 report.OwnershipEpoch = 10
if err := store.ReplaceRuntime(context.Background(), report, time.Minute); !errors.Is(err, ErrSnapshotMismatch) { if err := store.ReplaceRuntime(context.Background(), report, time.Minute); !errors.Is(err, ErrStaleReport) {
t.Fatalf("ReplaceRuntime(ahead epoch) error = %v, want ErrSnapshotMismatch", err) t.Fatalf("ReplaceRuntime(ahead epoch) error = %v, want ErrStaleReport", err)
} }
} }
@ -268,26 +154,3 @@ func registerRuntimeSession(t *testing.T, store *MemoryStore, sessionID string,
t.Fatalf("ReplaceSession(): %v", err) t.Fatalf("ReplaceSession(): %v", err)
} }
} }
func openControlSession(t *testing.T, store *MemoryStore, sessionID string, ttl time.Duration) {
t.Helper()
if err := store.OpenSession(context.Background(), Session{
WorkerID: "worker-a", InstanceID: "instance-a", SessionID: sessionID, Zone: "zone-a", ProtocolVersion: 1,
Labels: map[string]string{"region": "test"},
}, ttl); err != nil {
t.Fatalf("OpenSession(): %v", err)
}
}
func controlReference(version, epoch uint64, content string) SnapshotReference {
return SnapshotReference{
WorkerID: "worker-a", Version: version, OwnershipEpoch: epoch, Checksum: sha256.Sum256([]byte(content)),
}
}
func controlReport(now time.Time, sessionID string, sequence, version, epoch uint64) Report {
return Report{
WorkerID: "worker-a", SessionID: sessionID, Sequence: sequence, SnapshotVersion: version,
OwnershipEpoch: epoch, ObservedAt: now, Counters: []Counter{{ProxyID: "proxy-a", Active: 1}},
}
}

49
internal/domain/workerruntime/runtime.go
View File

@ -2,53 +2,26 @@ package workerruntime
import ( import (
"context" "context"
"crypto/sha256"
"errors" "errors"
"time" "time"
) )
var ( var (
ErrInvalidStore = errors.New("invalid worker runtime store") ErrInvalidStore = errors.New("invalid worker runtime store")
ErrInvalidSession = errors.New("invalid worker runtime session") ErrInvalidSession = errors.New("invalid worker runtime session")
ErrInvalidReport = errors.New("invalid worker runtime report") ErrInvalidReport = errors.New("invalid worker runtime report")
ErrInvalidQuery = errors.New("invalid worker runtime query") ErrInvalidQuery = errors.New("invalid worker runtime query")
ErrStaleSession = errors.New("stale worker runtime session") ErrStaleSession = errors.New("stale worker runtime session")
ErrStaleReport = errors.New("stale worker runtime report") ErrStaleReport = errors.New("stale worker runtime report")
ErrConflictingReport = errors.New("conflicting worker runtime report") ErrConflictingReport = errors.New("conflicting worker runtime report")
ErrInvalidSnapshotReference = errors.New("invalid worker snapshot reference")
ErrInvalidAcknowledgement = errors.New("invalid worker snapshot acknowledgement")
ErrSnapshotMismatch = errors.New("worker snapshot does not match acknowledged state")
ErrStaleSnapshotReference = errors.New("stale worker snapshot reference")
ErrConflictingSnapshotReference = errors.New("conflicting worker snapshot reference")
ErrStaleAcknowledgement = errors.New("stale worker snapshot acknowledgement")
) )
type Session struct { type Session struct {
WorkerID string WorkerID string
InstanceID string InstanceID string
SessionID string SessionID string
Zone string
ProtocolVersion uint32
Labels map[string]string
AckedSnapshotVersion uint64 AckedSnapshotVersion uint64
AckedOwnershipEpoch uint64 AckedOwnershipEpoch uint64
AckedChecksum [sha256.Size]byte
RuntimeEnabled bool
}
type SnapshotReference struct {
WorkerID string
Version uint64
OwnershipEpoch uint64
Checksum [sha256.Size]byte
}
type SnapshotAcknowledgement struct {
WorkerID string
SessionID string
Reference SnapshotReference
Applied bool
ErrorCode string
} }
type Counter struct { type Counter struct {
@ -88,14 +61,6 @@ type SessionWriter interface {
ReplaceSession(context.Context, Session, time.Duration) error ReplaceSession(context.Context, Session, time.Duration) error
} }
type ControlStore interface {
CurrentOwnershipEpoch(context.Context) (uint64, error)
OpenSession(context.Context, Session, time.Duration) error
RecordIssuedSnapshot(context.Context, SnapshotReference, time.Duration) error
AcknowledgeSnapshot(context.Context, SnapshotAcknowledgement, time.Duration) error
ReplaceRuntime(context.Context, Report, time.Duration) error
}
type ReportWriter interface { type ReportWriter interface {
ReplaceRuntime(context.Context, Report, time.Duration) error ReplaceRuntime(context.Context, Report, time.Duration) error
} }

138
internal/domain/workerruntime/validation.go
View File

@ -1,138 +0,0 @@
package workerruntime
import (
"crypto/sha256"
"encoding/json"
"regexp"
"sort"
"strings"
)
const (
maximumLabels = 32
maximumLabelKeyBytes = 64
maximumLabelValueBytes = 256
maximumLabelTotalBytes = 4 << 10
)
var identifierPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$`)
// ValidIdentifier accepts stable Worker, Session and Proxy identifiers.
func ValidIdentifier(value string) bool {
return identifierPattern.MatchString(value)
}
// NormalizeLabels validates and deep-copies the bounded Worker label set.
func NormalizeLabels(labels map[string]string) (map[string]string, error) {
if len(labels) > maximumLabels {
return nil, ErrInvalidSession
}
normalized := make(map[string]string, len(labels))
total := 0
for key, value := range labels {
if !ValidIdentifier(key) || len(key) > maximumLabelKeyBytes || value == "" ||
strings.TrimSpace(value) != value || strings.IndexByte(value, 0) >= 0 ||
len(value) > maximumLabelValueBytes {
return nil, ErrInvalidSession
}
total += len(key) + len(value)
if total > maximumLabelTotalBytes {
return nil, ErrInvalidSession
}
normalized[key] = value
}
return normalized, nil
}
// NormalizeSession prepares a new, not-yet-acknowledged session for storage.
func NormalizeSession(session Session) (Session, error) {
if !ValidIdentifier(session.WorkerID) || !ValidIdentifier(session.InstanceID) ||
!ValidIdentifier(session.SessionID) || !ValidIdentifier(session.Zone) ||
session.ProtocolVersion == 0 || session.AckedSnapshotVersion != 0 ||
session.AckedOwnershipEpoch != 0 || !checksumIsZero(session.AckedChecksum) ||
session.RuntimeEnabled {
return Session{}, ErrInvalidSession
}
labels, err := NormalizeLabels(session.Labels)
if err != nil {
return Session{}, err
}
session.Labels = labels
return session, nil
}
func NormalizeSnapshotReference(reference SnapshotReference) (SnapshotReference, error) {
if !ValidIdentifier(reference.WorkerID) || reference.Version == 0 || reference.OwnershipEpoch == 0 ||
checksumIsZero(reference.Checksum) {
return SnapshotReference{}, ErrInvalidSnapshotReference
}
return reference, nil
}
func NormalizeAcknowledgement(acknowledgement SnapshotAcknowledgement) (SnapshotAcknowledgement, error) {
if !ValidIdentifier(acknowledgement.WorkerID) || !ValidIdentifier(acknowledgement.SessionID) ||
(acknowledgement.ErrorCode != "" && !ValidIdentifier(acknowledgement.ErrorCode)) {
return SnapshotAcknowledgement{}, ErrInvalidAcknowledgement
}
reference, err := NormalizeSnapshotReference(acknowledgement.Reference)
if err != nil || reference.WorkerID != acknowledgement.WorkerID {
return SnapshotAcknowledgement{}, ErrInvalidAcknowledgement
}
acknowledgement.Reference = reference
return acknowledgement, 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 ||
report.SnapshotVersion == 0 || report.OwnershipEpoch == 0 || report.ObservedAt.IsZero() {
return Report{}, [sha256.Size]byte{}, ErrInvalidReport
}
normalized := report
normalized.ObservedAt = report.ObservedAt.UTC()
normalized.Counters = append([]Counter(nil), report.Counters...)
sort.Slice(normalized.Counters, func(left, right int) bool {
return normalized.Counters[left].ProxyID < normalized.Counters[right].ProxyID
})
seen := make(map[string]struct{}, len(normalized.Counters))
for _, counter := range normalized.Counters {
if !ValidIdentifier(counter.ProxyID) || counter.Active < 0 || counter.Reserved < 0 {
return Report{}, [sha256.Size]byte{}, ErrInvalidReport
}
if _, exists := seen[counter.ProxyID]; exists {
return Report{}, [sha256.Size]byte{}, ErrInvalidReport
}
seen[counter.ProxyID] = struct{}{}
}
payload, err := json.Marshal(normalized)
if err != nil {
return Report{}, [sha256.Size]byte{}, ErrInvalidReport
}
return normalized, sha256.Sum256(payload), nil
}
func checksumIsZero(checksum [sha256.Size]byte) bool {
return checksum == [sha256.Size]byte{}
}
func compareSnapshotTuple(left, right SnapshotReference) int {
switch {
case left.OwnershipEpoch < right.OwnershipEpoch:
return -1
case left.OwnershipEpoch > right.OwnershipEpoch:
return 1
case left.Version < right.Version:
return -1
case left.Version > right.Version:
return 1
default:
return 0
}
}
func sessionReference(session Session) SnapshotReference {
return SnapshotReference{
WorkerID: session.WorkerID, Version: session.AckedSnapshotVersion,
OwnershipEpoch: session.AckedOwnershipEpoch, Checksum: session.AckedChecksum,
}
}

63
internal/domain/workerruntime/validation_test.go
View File

@ -1,63 +0,0 @@
package workerruntime
import (
"crypto/sha256"
"errors"
"testing"
"time"
)
func TestNormalizeLabelsClonesAndBoundsValues(t *testing.T) {
source := map[string]string{"region": "cn-north"}
labels, err := NormalizeLabels(source)
if err != nil {
t.Fatalf("NormalizeLabels(): %v", err)
}
labels["region"] = "changed"
if source["region"] != "cn-north" {
t.Fatal("NormalizeLabels() aliases the source map")
}
if !ValidIdentifier("worker-a:1") || ValidIdentifier("worker a") || ValidIdentifier("") {
t.Fatal("ValidIdentifier() accepted or rejected an invalid value")
}
if _, err := NormalizeLabels(map[string]string{" region": "cn"}); !errors.Is(err, ErrInvalidSession) {
t.Fatalf("NormalizeLabels(invalid key) error = %v", err)
}
}
func TestNormalizeReportUsesStableCounterOrdering(t *testing.T) {
now := time.Date(2026, 7, 31, 9, 0, 0, 0, time.FixedZone("UTC+8", 8*60*60))
base := Report{
WorkerID: "worker-a", SessionID: "session-a", Sequence: 1, SnapshotVersion: 7, OwnershipEpoch: 3,
ObservedAt: now, Counters: []Counter{{ProxyID: "proxy-b", Active: 2}, {ProxyID: "proxy-a", Reserved: 1}},
}
normalized, digest, err := NormalizeReport(base)
if err != nil || normalized.ObservedAt.Location() != time.UTC || normalized.Counters[0].ProxyID != "proxy-a" {
t.Fatalf("NormalizeReport() = %+v, %x, %v", normalized, digest, err)
}
base.Counters[0], base.Counters[1] = base.Counters[1], base.Counters[0]
_, replayDigest, err := NormalizeReport(base)
if err != nil || digest != replayDigest {
t.Fatalf("NormalizeReport(reordered) digest = %x, %v; want %x", replayDigest, err, digest)
}
if _, _, err := NormalizeReport(Report{}); !errors.Is(err, ErrInvalidReport) {
t.Fatalf("NormalizeReport(invalid) error = %v", err)
}
}
func TestNormalizeSnapshotReferenceAndAcknowledgement(t *testing.T) {
reference := SnapshotReference{
WorkerID: "worker-a", Version: 7, OwnershipEpoch: 3, Checksum: sha256.Sum256([]byte("snapshot")),
}
if _, err := NormalizeSnapshotReference(reference); err != nil {
t.Fatalf("NormalizeSnapshotReference(): %v", err)
}
acknowledgement := SnapshotAcknowledgement{WorkerID: "worker-a", SessionID: "session-a", Reference: reference, ErrorCode: "apply_failed"}
if _, err := NormalizeAcknowledgement(acknowledgement); err != nil {
t.Fatalf("NormalizeAcknowledgement(): %v", err)
}
acknowledgement.Reference.WorkerID = "worker-b"
if _, err := NormalizeAcknowledgement(acknowledgement); !errors.Is(err, ErrInvalidAcknowledgement) {
t.Fatalf("NormalizeAcknowledgement(worker mismatch) error = %v", err)
}
}

62
scripts/generate-proto.ps1
View File

@ -1,62 +0,0 @@
param(
[string]$Protoc = "",
[string]$IncludePath = $env:PROTOC_INCLUDE,
[string]$OutputRoot = ""
)
$ErrorActionPreference = "Stop"
$repositoryRoot = Split-Path -Parent $PSScriptRoot
$protoRoot = Join-Path $repositoryRoot "api/proto"
$source = Join-Path $protoRoot "controlplane/v1/controlplane.proto"
$toolRoot = Join-Path $repositoryRoot ".tmp-proto/tools"
New-Item -ItemType Directory -Force -Path $toolRoot | Out-Null
if ([string]::IsNullOrWhiteSpace($Protoc)) {
$Protoc = & (Join-Path $PSScriptRoot "install-protoc.ps1")
}
$runningWindows = [System.Runtime.InteropServices.RuntimeInformation]::IsOSPlatform(
[System.Runtime.InteropServices.OSPlatform]::Windows
)
$suffix = if ($runningWindows) { ".exe" } else { "" }
$protocGenGo = Join-Path $toolRoot ("protoc-gen-go" + $suffix)
$protocGenGoGRPC = Join-Path $toolRoot ("protoc-gen-go-grpc" + $suffix)
go build -o $protocGenGo google.golang.org/protobuf/cmd/protoc-gen-go
if ($LASTEXITCODE -ne 0) { throw "build protoc-gen-go failed" }
go build -o $protocGenGoGRPC google.golang.org/grpc/cmd/protoc-gen-go-grpc
if ($LASTEXITCODE -ne 0) { throw "build protoc-gen-go-grpc failed" }
$protocCommand = Get-Command $Protoc -ErrorAction Stop
$protocVersion = (& $protocCommand.Source --version).Trim()
if ($protocVersion -ne "libprotoc 35.0") { throw "protoc 35.0 is required" }
if ([string]::IsNullOrWhiteSpace($IncludePath)) {
$installationRoot = Split-Path (Split-Path $protocCommand.Source -Parent) -Parent
$IncludePath = @(
(Join-Path $installationRoot "include"),
"/usr/include",
"/usr/local/include"
) |
Where-Object { Test-Path (Join-Path $_ "google/protobuf/timestamp.proto") } |
Select-Object -First 1
}
if ([string]::IsNullOrWhiteSpace($OutputRoot)) { $OutputRoot = $repositoryRoot }
if ([string]::IsNullOrWhiteSpace($IncludePath) -or
-not (Test-Path (Join-Path $IncludePath "google/protobuf/timestamp.proto"))) {
throw "protoc well-known type include directory was not found"
}
New-Item -ItemType Directory -Force -Path $OutputRoot | Out-Null
& $protocCommand.Source `
"--proto_path=$protoRoot" `
"--proto_path=$IncludePath" `
"--plugin=protoc-gen-go=$protocGenGo" `
"--plugin=protoc-gen-go-grpc=$protocGenGoGRPC" `
"--go_out=$OutputRoot" `
"--go_opt=module=proxy-pool" `
"--go-grpc_out=$OutputRoot" `
"--go-grpc_opt=module=proxy-pool" `
$source
if ($LASTEXITCODE -ne 0) { throw "protobuf Go generation failed" }

61
scripts/install-protoc.ps1
View File

@ -1,61 +0,0 @@
param()
$ErrorActionPreference = "Stop"
$repositoryRoot = Split-Path -Parent $PSScriptRoot
$runtime = [System.Runtime.InteropServices.RuntimeInformation]
$runningWindows = $runtime::IsOSPlatform([System.Runtime.InteropServices.OSPlatform]::Windows)
$runningLinux = $runtime::IsOSPlatform([System.Runtime.InteropServices.OSPlatform]::Linux)
if ($runtime::OSArchitecture -ne [System.Runtime.InteropServices.Architecture]::X64) {
throw "protoc bootstrap supports x64 only"
}
if ($runningWindows) {
$asset = "protoc-35.0-win64.zip"
$expectedHash = "d1cede9e308cc3eb072392af1c02ccae4bdd3d2f374ec2970dbd8cdfdaa91363"
$executableName = "protoc.exe"
}
elseif ($runningLinux) {
$asset = "protoc-35.0-linux-x86_64.zip"
$expectedHash = "a45cda0989c17dd950db55f6fbe1e5814c50fda08e87aa422980ac1f89dddbbc"
$executableName = "protoc"
}
else {
throw "protoc bootstrap supports Windows x64 and Linux x64 only"
}
$cacheDirectory = Join-Path $repositoryRoot ".tmp-proto/downloads"
$installationDirectory = Join-Path $repositoryRoot ".tmp-proto/protoc-35.0"
$archivePath = Join-Path $cacheDirectory $asset
$downloadURL = "https://github.com/protocolbuffers/protobuf/releases/download/v35.0/$asset"
New-Item -ItemType Directory -Force -Path $cacheDirectory | Out-Null
$needsDownload = -not (Test-Path -LiteralPath $archivePath)
if (-not $needsDownload) {
$actualHash = (Get-FileHash -LiteralPath $archivePath -Algorithm SHA256).Hash.ToLowerInvariant()
$needsDownload = $actualHash -ne $expectedHash
}
if ($needsDownload) {
Invoke-WebRequest -Uri $downloadURL -OutFile $archivePath
}
$actualHash = (Get-FileHash -LiteralPath $archivePath -Algorithm SHA256).Hash.ToLowerInvariant()
if ($actualHash -ne $expectedHash) {
throw "protoc archive SHA-256 mismatch"
}
New-Item -ItemType Directory -Force -Path $installationDirectory | Out-Null
Expand-Archive -LiteralPath $archivePath -DestinationPath $installationDirectory -Force
$protocPath = Join-Path $installationDirectory "bin/$executableName"
if (-not (Test-Path -LiteralPath $protocPath)) {
throw "protoc executable was not extracted"
}
$version = (& $protocPath --version).Trim()
if ($version -ne "libprotoc 35.0") {
throw "protoc version mismatch: $version"
}
(Resolve-Path -LiteralPath $protocPath).Path

31
scripts/verify-proto.ps1
View File

@ -1,5 +1,5 @@
param( param(
[string]$Protoc = "", [string]$Protoc = "protoc",
[string]$IncludePath = $env:PROTOC_INCLUDE, [string]$IncludePath = $env:PROTOC_INCLUDE,
[string]$OutputPath = "" [string]$OutputPath = ""
) )
@ -10,14 +10,7 @@ $repositoryRoot = Split-Path -Parent $PSScriptRoot
$protoRoot = Join-Path $repositoryRoot "api/proto" $protoRoot = Join-Path $repositoryRoot "api/proto"
$source = Join-Path $protoRoot "controlplane/v1/controlplane.proto" $source = Join-Path $protoRoot "controlplane/v1/controlplane.proto"
if ([string]::IsNullOrWhiteSpace($Protoc)) {
$Protoc = & (Join-Path $PSScriptRoot "install-protoc.ps1")
}
$protocCommand = Get-Command $Protoc -ErrorAction Stop $protocCommand = Get-Command $Protoc -ErrorAction Stop
$protocVersion = (& $protocCommand.Source --version).Trim()
if ($protocVersion -ne "libprotoc 35.0") {
throw "protoc 35.0 is required"
}
if ([string]::IsNullOrWhiteSpace($IncludePath)) { if ([string]::IsNullOrWhiteSpace($IncludePath)) {
$installationRoot = Split-Path (Split-Path $protocCommand.Source -Parent) -Parent $installationRoot = Split-Path (Split-Path $protocCommand.Source -Parent) -Parent
$candidates = @( $candidates = @(
@ -56,25 +49,3 @@ if ($descriptor.Length -le 0) {
throw "protoc produced an empty descriptor set" throw "protoc produced an empty descriptor set"
} }
Write-Host "descriptor: $($descriptor.FullName) ($($descriptor.Length) bytes)" Write-Host "descriptor: $($descriptor.FullName) ($($descriptor.Length) bytes)"
$outputRoot = Join-Path $repositoryRoot ".tmp-proto/generated"
& (Join-Path $PSScriptRoot "generate-proto.ps1") `
-Protoc $protocCommand.Source `
-IncludePath $IncludePath `
-OutputRoot $outputRoot
$generated = @(
"gen/controlplane/v1/controlplane.pb.go",
"gen/controlplane/v1/controlplane_grpc.pb.go"
)
foreach ($relative in $generated) {
$committed = Join-Path $repositoryRoot $relative
$candidate = Join-Path $outputRoot $relative
if (-not (Test-Path -LiteralPath $committed) -or
-not (Test-Path -LiteralPath $candidate) -or
-not [System.Linq.Enumerable]::SequenceEqual(
[System.IO.File]::ReadAllBytes($committed),
[System.IO.File]::ReadAllBytes($candidate))) {
throw "generated protobuf drift: $relative"
}
}

7
scripts/verify.ps1
View File

@ -22,7 +22,12 @@ if ($unformatted) {
Invoke-Step "go vet" { go vet ./... } Invoke-Step "go vet" { go vet ./... }
Invoke-Step "unit tests" { go test -timeout 60s ./... } Invoke-Step "unit tests" { go test -timeout 60s ./... }
Invoke-Step "protobuf contracts" { & (Join-Path $PSScriptRoot "verify-proto.ps1") } if (Get-Command protoc -ErrorAction SilentlyContinue) {
Invoke-Step "protobuf descriptor" { & (Join-Path $PSScriptRoot "verify-proto.ps1") }
}
else {
Write-Host "==> protobuf descriptor skipped: protoc is not installed"
}
if ((go env CGO_ENABLED) -eq "1") { if ((go env CGO_ENABLED) -eq "1") {
Invoke-Step "race tests" { go test -race -timeout 60s ./internal/... } Invoke-Step "race tests" { go test -race -timeout 60s ./internal/... }