feat: refresh worker snapshots before expiry
This commit is contained in:
youfak
parent
84ed10bd7a
commit
b9b323af9b
@ -55,7 +55,8 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费
|
||||
View 匹配请求，并在内存中按 Sequential、Random、Round Robin、Weighted 或 Least
|
||||
Connections 选择上游。无候选时支持 reject、受 `waitTimeout` 限制的本地容量等待，
|
||||
以及仍经过目标地址策略的 direct；`proxy-gateway` 通过独立控制面拨号地址维护
|
||||
Session，并仅在持有未过期 Snapshot 时 Ready；凭据材料只保留在当前节点内存 View。
|
||||
Session，并在每份 Snapshot 有效期的一半前接收版本递增的完整刷新；仅在持有未过期
|
||||
Snapshot 时 Ready，凭据材料只保留在当前节点内存 View。
|
||||
- **安全边界**：Gateway、Distribution 与 Admin 使用各自的认证语义，并支持
|
||||
CIDR、可信代理、严格请求解析和敏感信息最小化。
|
||||
|
||||
|
||||
@ -25,7 +25,9 @@ Controller 已实现并验证 `RegisterWorker`、`AcknowledgeSnapshot` 和
|
||||
监听允许明文 fixture 模式。单消息大小、并发流数和 gRPC keepalive 策略由
|
||||
`controlPlane` 配置限定。
|
||||
|
||||
`WatchSnapshots` 已在 Register 后发送与当前 ownership epoch 对应的基础完整快照，
|
||||
`WatchSnapshots` 已在 Register 后发送与当前 ownership epoch 对应的基础完整快照，并在
|
||||
每份快照有效期的一半前重新构建、下发版本递增的完整快照；这样 ownership、Routing 和
|
||||
凭据变化会在同一长连接内收敛，而无需等待有效期到达后重新注册。
|
||||
Gateway 校验后 ACK 并开始 Runtime 心跳。Controller 会从 Redis 的有界 Worker ownership
|
||||
索引构建已归属 Proxy 内容，并将租约到期收紧到 Proxy 的 `usable_until`。Proxy 引用的
|
||||
凭据材料按 `secret_ref + credential_version` 去重，随完整 Snapshot 经 mTLS 下发，仅保留在
|
||||
@ -108,11 +110,10 @@ Gateway 接收完整快照时必须拒绝缺失、格式错误或已到期的 `v
|
||||
保存在本地不可变视图。该整体期限到达后，调度直接按无候选处理，不再使用旧视图
|
||||
发起新的上游连接，也不查询 Redis 或 PostgreSQL 补偿。
|
||||
|
||||
Controller 只会下发尚未到期的完整快照，并在最近一次成功下发快照的
|
||||
`valid_until` 到达时结束 `WatchSnapshots` 流。Gateway 的 `SessionSupervisor` 会在流
|
||||
结束或可恢复控制面错误后按带 jitter 的有界退避重建 Register/Watch 会话；参数、认证
|
||||
和协议不兼容错误直接返回。新快照通过校验并原子替换前，旧视图仍按其整体有效期
|
||||
fail-closed。
|
||||
Controller 只会下发尚未到期的完整快照，并在每份快照有效期的一半前发送下一版完整
|
||||
快照。刷新构建失败、流结束或可恢复控制面错误时，Gateway 的 `SessionSupervisor` 会按
|
||||
带 jitter 的有界退避重建 Register/Watch 会话；参数、认证和协议不兼容错误直接返回。
|
||||
新快照通过校验并原子替换前，旧视图仍按其整体有效期 fail-closed。
|
||||
|
||||
Delta 声明 `base_version`。Worker 只有在本地版本恰好等于 base 且 checksum
|
||||
验证成功时才能应用；否则丢弃 Delta 并请求完整 Snapshot。构建在后台完成，
|
||||
|
||||
@ -22,7 +22,7 @@ go run ./cmd/proxy-controller -config CONFIG_FILE
|
||||
PostgreSQL 管理面迁移、Redis 活动池、Distribution/Admin 独立监听与优雅停机；
|
||||
Controller Metrics 独立监听、`/livez`、`/readyz` 和基础 Prometheus 运行时指标；
|
||||
Provider 自动补池、分布式配额、动态重载和 Admin 低基数统计已装配。Gateway 进程、
|
||||
Worker 控制面会话、Snapshot 凭据分发和 Snapshot 就绪探针已装配；Checker 的任务流与
|
||||
Worker 控制面会话、持续完整 Snapshot 刷新、凭据分发和 Snapshot 就绪探针已装配；Checker 的任务流与
|
||||
租约契约已接入控制面，但 Redis 共享调度、独立执行器与代表性负载验证仍在后续实施范围。
|
||||
|
||||
所有时间值使用 Go duration，例如 `500ms`、`30s`、`5m`。示例中的
|
||||
|
||||
@ -57,7 +57,8 @@ Provider、Pool、Routing、Distribution 在首版需要共享事务和一致性
|
||||
- 首条命中 Routing、Upstream 选择和 Proxy least-connections 选择。
|
||||
- 本地原子容量预留、建连、Active 计数和结果上报。
|
||||
- 本地不可变 Snapshot；热路径无数据库/Redis/Provider API。
|
||||
- 独立的控制面 SessionSupervisor；快照流到期或中断后以有界退避重建会话。
|
||||
- 独立的控制面 SessionSupervisor；Controller 在有效期一半前连续下发版本递增的完整
|
||||
快照，流中断后才以有界退避重建会话。
|
||||
- SSRF 与 DNS Rebinding 防护。
|
||||
|
||||
### 3.2 proxy-controller
|
||||
|
||||
10
findings.md
10
findings.md
@ -114,6 +114,16 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两
|
||||
维护该索引。Memory 单测、Redis 集成契约和配置/Reaper 单测覆盖恢复、阈值、延后与
|
||||
Drain/ACK 后清理。
|
||||
|
||||
## Worker Snapshot 刷新（2026-08-02）
|
||||
|
||||
- `OwnedSnapshotSource` 每次调用只构建一份权威完整快照；新增
|
||||
`RefreshingSnapshotSource` 以有效期一半为上限持续调用该窄接口，并传递最新版本和
|
||||
checksum。短于全局 `maxStaleAge` 的 Proxy/ownership 租约会进一步缩短下一次刷新间隔。
|
||||
- gRPC Handler 每次成功发送完整 Snapshot 都重置服务端到期计时器；流中刷新失败或中断时
|
||||
Gateway 使用既有 `SessionSupervisor` 退避重连。Gateway 本地 Store 对从完整 Snapshot
|
||||
消失且仍有 Active/Reserved 的 Proxy 已按 draining 继续上报，但 Controller 尚未具备
|
||||
自动 `BeginDrain`/`AcknowledgeDrain` 编排或 Worker 分配器，不能视为自动 Drain 闭环。
|
||||
|
||||
## Git 同步事实（2026-07-29）
|
||||
|
||||
- PostgreSQL 管理面基础文档已提交为 `7951c29`。
|
||||
|
||||
@ -60,9 +60,13 @@ func NewServer(controlPlane config.ControlPlane, service Service, options Server
|
||||
}
|
||||
}
|
||||
if snapshots == nil {
|
||||
snapshots, err = NewInitialSnapshotSource(service, controlPlane.MaxStaleAge.Value(), time.Now)
|
||||
initial, initialErr := NewInitialSnapshotSource(service, controlPlane.MaxStaleAge.Value(), time.Now)
|
||||
if initialErr != nil {
|
||||
return nil, fmt.Errorf("%w: build initial snapshot source: %v", ErrInvalidServer, initialErr)
|
||||
}
|
||||
snapshots, err = NewRefreshingSnapshotSource(initial, snapshotRefreshEvery(controlPlane.MaxStaleAge.Value()))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: build initial snapshot source: %v", ErrInvalidServer, err)
|
||||
return nil, fmt.Errorf("%w: build refreshing snapshot source: %v", ErrInvalidServer, err)
|
||||
}
|
||||
}
|
||||
serverOptions = append(serverOptions,
|
||||
|
||||
@ -156,7 +156,11 @@ func NewService(store workerruntime.ControlStore, options Options) (Service, err
|
||||
if err != nil {
|
||||
return nil, errors.Join(ErrInvalidCommand, err)
|
||||
}
|
||||
result.snapshots = source
|
||||
refreshed, refreshErr := NewRefreshingSnapshotSource(source, snapshotRefreshEvery(options.MaxStaleAge))
|
||||
if refreshErr != nil {
|
||||
return nil, errors.Join(ErrInvalidCommand, refreshErr)
|
||||
}
|
||||
result.snapshots = refreshed
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
@ -25,6 +25,113 @@ type SnapshotSource interface {
|
||||
Watch(context.Context, SnapshotWatchRequest) (<-chan *controlplanev1.WorkerSnapshot, error)
|
||||
}
|
||||
|
||||
// RefreshingSnapshotSource keeps a Worker snapshot stream current without
|
||||
// requiring the Gateway to reconnect after every validity window. It only
|
||||
// forwards complete snapshots, so Gateway delta handling remains fail-closed.
|
||||
type RefreshingSnapshotSource struct {
|
||||
source SnapshotSource
|
||||
refreshEvery time.Duration
|
||||
}
|
||||
|
||||
func NewRefreshingSnapshotSource(source SnapshotSource, refreshEvery time.Duration) (*RefreshingSnapshotSource, error) {
|
||||
if source == nil || refreshEvery <= 0 {
|
||||
return nil, ErrSnapshotsUnavailable
|
||||
}
|
||||
return &RefreshingSnapshotSource{source: source, refreshEvery: refreshEvery}, nil
|
||||
}
|
||||
|
||||
func snapshotRefreshEvery(validFor time.Duration) time.Duration {
|
||||
if validFor <= 0 {
|
||||
return 0
|
||||
}
|
||||
if half := validFor / 2; half > 0 {
|
||||
return half
|
||||
}
|
||||
return validFor
|
||||
}
|
||||
|
||||
func (source *RefreshingSnapshotSource) Watch(
|
||||
ctx context.Context,
|
||||
request SnapshotWatchRequest,
|
||||
) (<-chan *controlplanev1.WorkerSnapshot, error) {
|
||||
if source == nil || source.source == nil || ctx == nil {
|
||||
return nil, ErrSnapshotsUnavailable
|
||||
}
|
||||
first, err := source.next(ctx, request)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
updates := make(chan *controlplanev1.WorkerSnapshot, 1)
|
||||
updates <- first
|
||||
go source.refresh(ctx, updates, request, first)
|
||||
return updates, nil
|
||||
}
|
||||
|
||||
func (source *RefreshingSnapshotSource) refresh(
|
||||
ctx context.Context,
|
||||
updates chan<- *controlplanev1.WorkerSnapshot,
|
||||
request SnapshotWatchRequest,
|
||||
current *controlplanev1.WorkerSnapshot,
|
||||
) {
|
||||
defer close(updates)
|
||||
version := current.GetVersion()
|
||||
checksum := append([]byte(nil), current.GetChecksum()...)
|
||||
timer := time.NewTimer(source.delayFor(current))
|
||||
defer timer.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-timer.C:
|
||||
request.LastAppliedVersion = version
|
||||
request.LastChecksum = append(request.LastChecksum[:0], checksum...)
|
||||
next, err := source.next(ctx, request)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case updates <- next:
|
||||
version = next.GetVersion()
|
||||
checksum = append(checksum[:0], next.GetChecksum()...)
|
||||
timer.Reset(source.delayFor(next))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (source *RefreshingSnapshotSource) next(
|
||||
ctx context.Context,
|
||||
request SnapshotWatchRequest,
|
||||
) (*controlplanev1.WorkerSnapshot, error) {
|
||||
stream, err := source.source.Watch(ctx, request)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
case snapshot, ok := <-stream:
|
||||
if !ok || snapshot == nil || snapshot.GetVersion() <= request.LastAppliedVersion {
|
||||
return nil, ErrSnapshotsUnavailable
|
||||
}
|
||||
return snapshot, nil
|
||||
}
|
||||
}
|
||||
|
||||
func (source *RefreshingSnapshotSource) delayFor(snapshot *controlplanev1.WorkerSnapshot) time.Duration {
|
||||
delay := source.refreshEvery
|
||||
if snapshot == nil || snapshot.GetGeneratedAt() == nil || snapshot.GetValidUntil() == nil {
|
||||
return delay
|
||||
}
|
||||
lifetime := snapshot.GetValidUntil().AsTime().Sub(snapshot.GetGeneratedAt().AsTime())
|
||||
if half := lifetime / 2; half > 0 && half < delay {
|
||||
return half
|
||||
}
|
||||
return delay
|
||||
}
|
||||
|
||||
type OwnershipEpochReader interface {
|
||||
CurrentOwnershipEpoch(context.Context) (uint64, error)
|
||||
}
|
||||
@ -73,3 +180,4 @@ func (source *InitialSnapshotSource) Watch(ctx context.Context, request Snapshot
|
||||
}
|
||||
|
||||
var _ SnapshotSource = (*InitialSnapshotSource)(nil)
|
||||
var _ SnapshotSource = (*RefreshingSnapshotSource)(nil)
|
||||
|
||||
@ -3,9 +3,11 @@ package worker
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"google.golang.org/protobuf/types/known/timestamppb"
|
||||
controlplanev1 "proxy-pool/gen/controlplane/v1"
|
||||
"proxy-pool/internal/controlplane/snapshotwire"
|
||||
ownershipDomain "proxy-pool/internal/domain/ownership"
|
||||
@ -35,6 +37,47 @@ func TestInitialSnapshotSourceIssuesNextFullSnapshot(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRefreshingSnapshotSourceIssuesSequentialFullSnapshots(t *testing.T) {
|
||||
base := &refreshingSnapshotSourceStub{}
|
||||
source, err := NewRefreshingSnapshotSource(base, 10*time.Millisecond)
|
||||
if err != nil {
|
||||
t.Fatalf("NewRefreshingSnapshotSource() = %v", err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
updates, err := source.Watch(ctx, SnapshotWatchRequest{
|
||||
WorkerID: "worker-a", SessionID: "session-a", LastAppliedVersion: 4, LastChecksum: []byte{4},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Watch() = %v", err)
|
||||
}
|
||||
first := receiveSnapshot(t, updates)
|
||||
second := receiveSnapshot(t, updates)
|
||||
if first.GetVersion() != 5 || second.GetVersion() != 6 || second.GetVersion() <= first.GetVersion() {
|
||||
t.Fatalf("refreshed versions = (%d, %d)", first.GetVersion(), second.GetVersion())
|
||||
}
|
||||
if calls := base.Calls(); len(calls) < 2 || calls[0] != 4 || calls[1] != 5 {
|
||||
t.Fatalf("base requested versions = %v", calls)
|
||||
}
|
||||
if checksums := base.Checksums(); len(checksums) < 2 || string(checksums[0]) != string([]byte{4}) ||
|
||||
string(checksums[1]) != string([]byte{5}) {
|
||||
t.Fatalf("base requested checksums = %v", checksums)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRefreshingSnapshotSourceRefreshesBeforeShortSnapshotExpiry(t *testing.T) {
|
||||
source, err := NewRefreshingSnapshotSource(&refreshingSnapshotSourceStub{}, 2*time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("NewRefreshingSnapshotSource() = %v", err)
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
if delay := source.delayFor(&controlplanev1.WorkerSnapshot{
|
||||
GeneratedAt: timestamppb.New(now), ValidUntil: timestamppb.New(now.Add(100 * time.Millisecond)),
|
||||
}); delay != 50*time.Millisecond {
|
||||
t.Fatalf("delayFor(short-lived snapshot) = %s, want 50ms", delay)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOwnedSnapshotSourceBuildsBoundedProxySnapshot(t *testing.T) {
|
||||
now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC)
|
||||
expiresAt := now.Add(10 * time.Minute)
|
||||
@ -167,6 +210,57 @@ type epochReaderStub struct {
|
||||
err error
|
||||
}
|
||||
|
||||
type refreshingSnapshotSourceStub struct {
|
||||
mu sync.Mutex
|
||||
calls []uint64
|
||||
checksums [][]byte
|
||||
}
|
||||
|
||||
func (source *refreshingSnapshotSourceStub) Watch(
|
||||
_ context.Context,
|
||||
request SnapshotWatchRequest,
|
||||
) (<-chan *controlplanev1.WorkerSnapshot, error) {
|
||||
source.mu.Lock()
|
||||
source.calls = append(source.calls, request.LastAppliedVersion)
|
||||
source.checksums = append(source.checksums, append([]byte(nil), request.LastChecksum...))
|
||||
source.mu.Unlock()
|
||||
updates := make(chan *controlplanev1.WorkerSnapshot, 1)
|
||||
updates <- &controlplanev1.WorkerSnapshot{
|
||||
Version: request.LastAppliedVersion + 1, Checksum: []byte{byte(request.LastAppliedVersion + 1)},
|
||||
}
|
||||
return updates, nil
|
||||
}
|
||||
|
||||
func (source *refreshingSnapshotSourceStub) Calls() []uint64 {
|
||||
source.mu.Lock()
|
||||
defer source.mu.Unlock()
|
||||
return append([]uint64(nil), source.calls...)
|
||||
}
|
||||
|
||||
func (source *refreshingSnapshotSourceStub) Checksums() [][]byte {
|
||||
source.mu.Lock()
|
||||
defer source.mu.Unlock()
|
||||
checksums := make([][]byte, len(source.checksums))
|
||||
for index := range source.checksums {
|
||||
checksums[index] = append([]byte(nil), source.checksums[index]...)
|
||||
}
|
||||
return checksums
|
||||
}
|
||||
|
||||
func receiveSnapshot(t *testing.T, updates <-chan *controlplanev1.WorkerSnapshot) *controlplanev1.WorkerSnapshot {
|
||||
t.Helper()
|
||||
select {
|
||||
case snapshot, ok := <-updates:
|
||||
if !ok || snapshot == nil {
|
||||
t.Fatal("snapshot updates closed before next full snapshot")
|
||||
}
|
||||
return snapshot
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("timed out waiting for snapshot")
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func (reader epochReaderStub) CurrentOwnershipEpoch(context.Context) (uint64, error) {
|
||||
return reader.epoch, reader.err
|
||||
}
|
||||
|
||||
@ -15,6 +15,10 @@
|
||||
- Redis 集成标签测试在未设置 `PROXY_POOL_TEST_REDIS_URL` 时现能正确跳过所有契约
|
||||
子测试；`go test ./...`、`go test -tags=integration ./internal/adapters/redisactivity`
|
||||
与 `scripts/verify.ps1` 已通过，race 仍由 Linux CI 执行。
|
||||
- Worker 控制面改为持续完整 Snapshot 发布：Controller 在每份 Snapshot 有效期的一半
|
||||
前重建并下发递增版本，Gateway 在同一长连接内原子替换视图；构建失败或流中断时才由
|
||||
SessionSupervisor 按原有退避重连。该机制复用 `RefreshingSnapshotSource`，不降低
|
||||
Delta 的 fail-closed 校验，也不把自动 Drain/ACK 记为已完成。
|
||||
|
||||
## 2026-07-30
|
||||
|
||||
|
||||
@ -40,8 +40,8 @@
|
||||
14. [已完成] 收敛健康故障生命周期：将 `UNHEALTHY` 的持续时长纳入权威活动池状态，
|
||||
通过 Controller 后台的有界回收清理无 Worker ownership 的持续异常 Proxy；拥有
|
||||
Worker 的候选延后到既有 Drain/ACK 清除所有权后再处理。
|
||||
15. [进行中] 收敛 Worker 发布生命周期：补齐权威 Proxy/Routing 的增量发布和
|
||||
自动 Drain 编排，使配置停用、健康淘汰与 Snapshot 撤销形成可观测闭环。
|
||||
15. [进行中] 收敛 Worker 发布生命周期：已完成权威 Proxy/Routing 的持续完整快照
|
||||
刷新；后续补齐自动 Drain 编排，使配置停用、健康淘汰与 Snapshot 撤销形成可观测闭环。
|
||||
|
||||
## 串并行关系
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user