From 1729a5c868e3f2dbdf6a51ba493c6cba09d3a1f6 Mon Sep 17 00:00:00 2001 From: youfak Date: Sun, 2 Aug 2026 12:21:50 +0800 Subject: [PATCH] feat: finalize drains from acknowledged runtime reports --- docs/adr/005-redis-activity-pool.md | 6 +- docs/api/control-plane.md | 8 +- findings.md | 11 ++- .../ownership_integration_test.go | 72 ++++++++++++++ internal/adapters/redisactivity/runtime.go | 13 ++- .../redisactivity/scripts/runtime.lua | 95 +++++++++++++++++++ progress.md | 8 +- task_plan.md | 4 +- 8 files changed, 202 insertions(+), 15 deletions(-) diff --git a/docs/adr/005-redis-activity-pool.md b/docs/adr/005-redis-activity-pool.md index 3d57244..4cf528c 100644 --- a/docs/adr/005-redis-activity-pool.md +++ b/docs/adr/005-redis-activity-pool.md @@ -196,8 +196,10 @@ Proxy 继续以 `draining=true` 上报，直到 Active/Reserved 同时归零。 Drain Ticket 会绑定“已签发给当前 session、且不早于所需 epoch 的完整排除快照”引用， 包括 version、epoch 和 checksum。 -后续会在 Runtime 替换 Lua 事务中同时验证 Snapshot ACK、Ticket 屏障和零计数，避免 -拆分为读 Runtime 再释放所有权产生竞态。 +Runtime 替换 Lua 事务会同时验证当前 session ACK、Ticket 屏障与该 Proxy 的零计数， +仅在完整稀疏报告中 Active/Reserved 都为零时释放所有权、删除 Ticket 和 Worker Drain +索引，并在记录仍为 AVAILABLE 时恢复可分配索引。ACK 或屏障不匹配、报告中仍有计数、 +或 session 已替换时都保留 Drain；这样避免拆分为读 Runtime 再释放所有权产生竞态。 Redis Adapter 在单个 `{activity}` 原子边界内维护 Worker session 和运行态报告： diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index 25775b4..48480a4 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -139,8 +139,12 @@ Delta 声明 `base_version`。Worker 只有在本地版本恰好等于 base 且 Drain Ticket，并推进全局 ownership epoch，促使下一份完整 Snapshot 撤销该 Proxy。Ticket 按 Worker 有界读取，供 Controller 在确认完整 Snapshot 确实不含该 Proxy 后绑定快照屏障。 Worker Handler 先登记该 Snapshot 引用，再将 Ticket 绑定到 session、version、epoch 与 -checksum。当前实现尚未把该屏障和 Runtime 报告放入同一最终确认事务，因此不会自动释放 -所有权；`AcknowledgeDrain` 仍是已有的显式完成原语。 +checksum。后续完整 Runtime 替换在同一 Redis Lua 事务中检查：Ticket 屏障是否属于当前 +session、该 session 是否已 ACK 不早于屏障的快照、以及本次完整稀疏报告中该 Proxy 的 +Active/Reserved 是否均为零（缺失项按零）。三者同时成立才清除 owner、Ticket 与 +Worker Drain 索引，并在记录仍为 AVAILABLE 时恢复可分配索引；因此不存在先读 Runtime +再释放所有权的竞争窗口。`AcknowledgeDrain` 保留为显式完成原语，用于不经 Runtime +上报的受控维护流程。 Worker 崩溃时必须等待所有权 epoch/有效期失效后再转移，避免双主。Proto 中 `ReportRuntimeResponse.revoke_proxy_ids` 是加速 Drain 的控制信号，不绕过 diff --git a/findings.md b/findings.md index 45c60c1..4a1e6c6 100644 --- a/findings.md +++ b/findings.md @@ -121,8 +121,9 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两 checksum。短于全局 `maxStaleAge` 的 Proxy/ownership 租约会进一步缩短下一次刷新间隔。 - gRPC Handler 每次成功发送完整 Snapshot 都重置服务端到期计时器；流中刷新失败或中断时 Gateway 使用既有 `SessionSupervisor` 退避重连。Gateway 本地 Store 对从完整 Snapshot - 消失且仍有 Active/Reserved 的 Proxy 已按 draining 继续上报，但 Controller 尚未具备 - 自动 `BeginDrain`/`AcknowledgeDrain` 编排或 Worker 分配器，不能视为自动 Drain 闭环。 + 消失且仍有 Active/Reserved 的 Proxy 已按 draining 继续上报。Redis Runtime 替换已在当前 + session ACK、Ticket 屏障和完整零计数同时成立时自动释放 owner；Reaper 和配置停用尚未 + 调用 `BeginDrain`，因此全链路的 Drain 触发编排仍未完成。 - Snapshot 的 `version` 是同一 Worker 流的连续序列，`ownership_epoch` 是独立且只能前进 的权威栅栏；epoch 变化不重置 version。此前 Gateway 错把 epoch 前进要求为 version=1， 与 Controller 的 `lastAppliedVersion+1` 生成规则冲突，现已用连续版本规则统一。 @@ -133,9 +134,9 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两 撤销”与“Gateway 已收到排除该 Proxy 的完整快照”。现已增加按 Worker 有界读取的持久化 Ticket，并以 RequiredSnapshotEpoch 强制下一份权威快照至少跨过 Drain 操作。Handler 在 Snapshot 引用登记成功后才绑定屏障，绑定内容包含 session、version、epoch 与 checksum。 -- Ticket 不能单独成为释放依据。最终清理必须把 Ticket 屏障、当前 session ACK 和完整 - Runtime 替换中的零计数置于同一 Redis Lua 原子边界；`ReadRuntime` 后再调用 - `AcknowledgeDrain` 会保留竞争窗口。 +- Ticket 不能单独成为释放依据。`replace_report` Lua 已把 Ticket 屏障、当前 session ACK + 和完整 Runtime 替换中的零计数置于同一原子边界；只有报告缺失该 Proxy 或其 + Active/Reserved 均为零时，才删除 owner/Ticket/Drain 索引并恢复可分配状态。 ## Git 同步事实（2026-07-29） diff --git a/internal/adapters/redisactivity/ownership_integration_test.go b/internal/adapters/redisactivity/ownership_integration_test.go index 000414d..62e0127 100644 --- a/internal/adapters/redisactivity/ownership_integration_test.go +++ b/internal/adapters/redisactivity/ownership_integration_test.go @@ -4,6 +4,7 @@ package redisactivity import ( "context" + "crypto/sha256" "errors" "fmt" "sync" @@ -14,6 +15,7 @@ import ( extractionDomain "proxy-pool/internal/domain/extraction" ownershipDomain "proxy-pool/internal/domain/ownership" proxyDomain "proxy-pool/internal/domain/proxy" + "proxy-pool/internal/domain/workerruntime" ) func TestRedisOwnershipLifecycle(t *testing.T) { @@ -149,6 +151,76 @@ func TestRedisOwnershipRejectsInvalidDrainAndStaleAssignment(t *testing.T) { } } +func TestRedisRuntimeFinalizesAcknowledgedBoundDrain(t *testing.T) { + fixture := newRedisTestFixture(t) + now := redisTestNow() + seedRedisAvailable(t, fixture.Adapter, "provider-a", now, now.Add(time.Second), 2*time.Minute, + testProxy("proxy-a", "192.0.2.10")) + assignment, err := fixture.Adapter.Assign(context.Background(), now.Add(2*time.Second), "proxy-a", "worker-a", time.Minute) + if err != nil { + t.Fatalf("Assign(): %v", err) + } + if _, err := fixture.Adapter.BeginDrain(context.Background(), "proxy-a", "worker-a", assignment.Epoch); err != nil { + t.Fatalf("BeginDrain(): %v", err) + } + pending, err := fixture.Adapter.PendingDrains(context.Background(), "worker-a", 1) + if err != nil || len(pending) != 1 { + t.Fatalf("PendingDrains() = %+v, %v", pending, err) + } + checksum := sha256.Sum256([]byte("drain-snapshot")) + reference := workerruntime.SnapshotReference{ + WorkerID: "worker-a", Version: 1, OwnershipEpoch: pending[0].RequiredSnapshotEpoch, Checksum: checksum, + } + if err := fixture.Adapter.OpenSession(context.Background(), workerruntime.Session{ + WorkerID: "worker-a", InstanceID: "instance-a", SessionID: "session-a", Zone: "zone-a", ProtocolVersion: 1, + }, time.Minute); err != nil { + t.Fatalf("OpenSession(): %v", err) + } + if err := fixture.Adapter.RecordIssuedSnapshot(context.Background(), "session-a", reference, time.Minute); err != nil { + t.Fatalf("RecordIssuedSnapshot(): %v", err) + } + pending[0].Barrier = ownershipDomain.SnapshotBarrier{ + SessionID: "session-a", Version: reference.Version, OwnershipEpoch: reference.OwnershipEpoch, Checksum: checksum, + } + if err := fixture.Adapter.BindDrainBarrier(context.Background(), pending[0]); err != nil { + t.Fatalf("BindDrainBarrier(): %v", err) + } + if err := fixture.Adapter.AcknowledgeSnapshot(context.Background(), workerruntime.SnapshotAcknowledgement{ + WorkerID: "worker-a", SessionID: "session-a", Reference: reference, Applied: true, + }, time.Minute); err != nil { + t.Fatalf("AcknowledgeSnapshot(): %v", err) + } + if err := fixture.Adapter.ReplaceRuntime(context.Background(), workerruntime.Report{ + WorkerID: "worker-a", SessionID: "session-a", Sequence: 1, + SnapshotVersion: reference.Version, OwnershipEpoch: reference.OwnershipEpoch, ObservedAt: now.Add(3 * time.Second), + Counters: []workerruntime.Counter{{ProxyID: "proxy-a", Active: 1, Draining: true}}, + }, time.Minute); err != nil { + t.Fatalf("ReplaceRuntime(active drain): %v", err) + } + if current, ok, err := fixture.Adapter.Get(context.Background(), "proxy-a"); err != nil || !ok || !current.Draining { + t.Fatalf("Get(active drain) = %+v, %t, %v", current, ok, err) + } + if err := fixture.Adapter.ReplaceRuntime(context.Background(), workerruntime.Report{ + WorkerID: "worker-a", SessionID: "session-a", Sequence: 2, + SnapshotVersion: reference.Version, OwnershipEpoch: reference.OwnershipEpoch, ObservedAt: now.Add(4 * time.Second), + }, time.Minute); err != nil { + t.Fatalf("ReplaceRuntime(drained): %v", err) + } + if current, ok, err := fixture.Adapter.Get(context.Background(), "proxy-a"); err != nil || ok { + t.Fatalf("Get(after automatic drain) = %+v, %t, %v", current, ok, err) + } + if pending, err := fixture.Adapter.PendingDrains(context.Background(), "worker-a", 1); err != nil || len(pending) != 0 { + t.Fatalf("PendingDrains(after automatic drain) = %+v, %v", pending, err) + } + result, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{ + RequestID: "req-after-auto-drain", ClientID: "client-a", Requested: 1, + Fulfillment: extractionDomain.Partial, Now: now.Add(5 * time.Second), + }) + if err != nil || result.Returned != 1 || result.Items[0].ID != "proxy-a" { + t.Fatalf("Extract(after automatic drain) = %+v, %v", result, err) + } +} + func TestRedisOwnershipExpireIsLimitedAndAllowsTakeover(t *testing.T) { fixture := newRedisTestFixture(t) now := redisTestNow() diff --git a/internal/adapters/redisactivity/runtime.go b/internal/adapters/redisactivity/runtime.go index 1e1ae37..1553ac3 100644 --- a/internal/adapters/redisactivity/runtime.go +++ b/internal/adapters/redisactivity/runtime.go @@ -454,7 +454,8 @@ func (a *Adapter) runRuntime( a.keys.workerSessions, a.keys.workerSessionExpiry, a.keys.workerSnapshots, a.keys.workerSnapshotExpiry, a.keys.workerRuntime, a.keys.workerRuntimeExpiry, a.keys.owners, a.keys.epoch, - a.keys.workerOutcomes, + a.keys.workerOutcomes, a.keys.records, a.keys.available, a.keys.ownerExpiry, + a.keys.drainTickets, a.keys.workerDraining(runtimeWorkerID(payload)), }, operation, ttlMS, a.options.CleanupLimit, string(payload), digest) if err != nil { return runtimeScriptReply{}, err @@ -466,6 +467,16 @@ func (a *Adapter) runRuntime( return reply, nil } +func runtimeWorkerID(payload []byte) string { + var value struct { + WorkerID string `json:"workerId"` + } + if json.Unmarshal(payload, &value) != nil { + return "" + } + return value.WorkerID +} + func validateRuntimeCall(ctx context.Context, adapter *Adapter) error { if ctx == nil || adapter == nil { return workerruntime.ErrInvalidStore diff --git a/internal/adapters/redisactivity/scripts/runtime.lua b/internal/adapters/redisactivity/scripts/runtime.lua index 9c16d91..f72edaf 100644 --- a/internal/adapters/redisactivity/scripts/runtime.lua +++ b/internal/adapters/redisactivity/scripts/runtime.lua @@ -7,6 +7,11 @@ local runtime_expiry_key = KEYS[6] local owners_key = KEYS[7] local epoch_key = KEYS[8] local outcomes_key = KEYS[9] +local records_key = KEYS[10] +local available_key = KEYS[11] +local owner_expiry_key = KEYS[12] +local drain_tickets_key = KEYS[13] +local worker_draining_key = KEYS[14] local operation = ARGV[1] local ttl_ms = tonumber(ARGV[2]) @@ -130,6 +135,77 @@ local function valid_owner(value, worker_id, ownership_epoch, now) type(value.expiresAtMs) == 'number' and value.expiresAtMs > now end +local function touch(key, expires_at_ms) + if redis.call('EXISTS', key) == 0 then + return + end + local current = redis.call('PEXPIRETIME', key) + if current < expires_at_ms then + redis.call('PEXPIREAT', key, expires_at_ms) + end +end + +local function add_available(id, record, at_ms) + local usable_until_ms = record and tonumber(record.usableUntilMs) + if not usable_until_ms or record.state ~= 'AVAILABLE' or usable_until_ms <= at_ms then + return + end + redis.call('ZADD', available_key, usable_until_ms, id) + touch(available_key, tonumber(record.expiresAtMs)) + if type(record.indexKeys) == 'table' then + for _, index_key in ipairs(record.indexKeys) do + if type(index_key) == 'string' and index_key ~= '' then + redis.call('ZADD', index_key, usable_until_ms, id) + touch(index_key, tonumber(record.expiresAtMs)) + end + end + end +end + +local function valid_drain_ticket(ticket, worker_id) + return ticket and ticket.version == 1 and type(ticket.proxyId) == 'string' and ticket.proxyId ~= '' and + ticket.workerId == worker_id and type(ticket.assignmentEpoch) == 'number' and ticket.assignmentEpoch > 0 and + type(ticket.requiredSnapshotEpoch) == 'number' and ticket.requiredSnapshotEpoch > ticket.assignmentEpoch and + type(ticket.workerIndexKey) == 'string' and ticket.workerIndexKey ~= '' and + type(ticket.sessionId) == 'string' and ticket.sessionId ~= '' and + type(ticket.snapshotVersion) == 'number' and ticket.snapshotVersion > 0 and + type(ticket.snapshotOwnershipEpoch) == 'number' and ticket.snapshotOwnershipEpoch >= ticket.requiredSnapshotEpoch and + type(ticket.snapshotChecksum) == 'string' and string.len(ticket.snapshotChecksum) == 64 and + string.match(ticket.snapshotChecksum, '^[0-9a-f]+$') ~= nil +end + +local function clear_finalized_drain(proxy_id, owner, ticket, at_ms) + local record = decode_table(redis.call('HGET', records_key, proxy_id)) + if record and record.ownerWorkerId == owner.workerId then + record.ownerWorkerId = nil + redis.call('HSET', records_key, proxy_id, cjson.encode(record)) + add_available(proxy_id, record, at_ms) + end + redis.call('HDEL', owners_key, proxy_id) + redis.call('ZREM', owner_expiry_key, proxy_id) + redis.call('HDEL', drain_tickets_key, proxy_id) + redis.call('ZREM', ticket.workerIndexKey, proxy_id) +end + +local function ticket_acknowledged(session, ticket) + if session.sessionId ~= ticket.sessionId then + return false + end + local barrier = { + ownershipEpoch = tostring(ticket.snapshotOwnershipEpoch), + snapshotVersion = tostring(ticket.snapshotVersion), + } + local acknowledged = { + ownershipEpoch = session.ackedOwnershipEpoch, + snapshotVersion = session.ackedSnapshotVersion, + } + local ordering = compare_reference(acknowledged, barrier) + if ordering < 0 then + return false + end + return ordering ~= 0 or session.ackedChecksum == ticket.snapshotChecksum +end + local now = now_ms() cleanup(now) @@ -398,6 +474,7 @@ if operation == 'replace_report' then end end local seen = {} + local counters = {} for _, counter in pairs(report.counters) do if type(counter) ~= 'table' or type(counter.proxyId) ~= 'string' or counter.proxyId == '' or type(counter.active) ~= 'number' or counter.active < 0 or counter.active ~= math.floor(counter.active) or @@ -406,6 +483,7 @@ if operation == 'replace_report' then return reply('invalid') end seen[counter.proxyId] = true + counters[counter.proxyId] = counter local owner = decode_table(redis.call('HGET', owners_key, counter.proxyId)) if not valid_owner(owner, report.workerId, report.ownershipEpoch, now) then return reply('stale') @@ -418,6 +496,23 @@ if operation == 'replace_report' then session.expiresAtMs = now + ttl_ms redis.call('HSET', sessions_key, report.workerId, cjson.encode(session)) redis.call('ZADD', session_expiry_key, session.expiresAtMs, report.workerId) + + local ticket_ids = redis.call('ZRANGE', worker_draining_key, 0, cleanup_limit - 1) + for _, proxy_id in ipairs(ticket_ids) do + local ticket = decode_table(redis.call('HGET', drain_tickets_key, proxy_id)) + local owner = decode_table(redis.call('HGET', owners_key, proxy_id)) + if not ticket then + redis.call('ZREM', worker_draining_key, proxy_id) + elseif valid_drain_ticket(ticket, report.workerId) and ticket.workerIndexKey == worker_draining_key and + owner and owner.draining == true and owner.workerId == report.workerId and + tonumber(owner.epoch) == tonumber(ticket.assignmentEpoch) and + valid_owner(owner, report.workerId, report.ownershipEpoch, now) and ticket_acknowledged(session, ticket) then + local counter = counters[proxy_id] + if not counter or (counter.active == 0 and counter.reserved == 0) then + clear_finalized_drain(proxy_id, owner, ticket, now) + end + end + end return reply('ok') end diff --git a/progress.md b/progress.md index d64574d..6dc21d3 100644 --- a/progress.md +++ b/progress.md @@ -11,14 +11,14 @@ - 已完成持续 `UNHEALTHY` 后的安全淘汰：首次异常时间写入 Redis TTL 活动池， `check.unhealthyRemoveAfter` 按 Upstream 生效；Controller 通过公用、有界的 Reaper 仅回收未分配代理。仍持有 Worker ownership 的候选会延后，待既有 - Drain/ACK 清空 ownership 后再删除，避免影响仍在 Gateway Snapshot 中使用的代理。 + Drain 收尾清空 ownership 后再删除，避免影响仍在 Gateway Snapshot 中使用的代理。 - 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 记为已完成。 + Delta 的 fail-closed 校验。 - Health Scheduler 现在先建立 EGRESS 与 Routing TARGET 的独立 due 引用，再投递 BASIC 任务；在单项小批次下，短 TTL Proxy 不会因 BASIC 先出队而错过首次出口/目标探测。 - 修复 Snapshot epoch/version 栅栏：Worker 流的 `version` 在 ownership epoch 前进时 @@ -28,7 +28,9 @@ epoch 与所需 Snapshot epoch 组成的待绑定 Ticket；Redis 使用 `drain-tickets` 与按 Worker 的 `worker-draining` 索引，ACK、过期和硬删除会清理 Ticket。Worker Handler 在持久登记完整 Snapshot 引用后，核验代理确实不在完整视图内，再绑定 Ticket 的 - session/version/epoch/checksum 屏障；ACK 与 Runtime 零计数的原子自动完成仍在后续切片。 + session/version/epoch/checksum 屏障；同一 Redis Runtime 替换事务验证当前 session ACK、 + Ticket 屏障和完整报告零计数后自动删除 owner/Ticket/Drain 索引并恢复 AVAILABLE 代理的 + 可分配索引，避免读写分离竞态。 ## 2026-07-30 diff --git a/task_plan.md b/task_plan.md index b6365f1..b52434f 100644 --- a/task_plan.md +++ b/task_plan.md @@ -42,8 +42,8 @@ Worker 的候选延后到既有 Drain/ACK 清除所有权后再处理。 15. [进行中] 收敛 Worker 发布生命周期：已完成权威 Proxy/Routing 的持续完整快照 刷新，以及 Drain 发起时的持久化 Ticket、Worker 待绑定索引、ownership epoch 推进和 - 完整排除 Snapshot 屏障绑定；后续在 ACK 与 Runtime 零计数之间补齐原子自动完成编排，使配置停用、 - 健康淘汰与 Snapshot 撤销形成可观测闭环。 + 完整排除 Snapshot 屏障绑定；已在 ACK 与 Runtime 零计数之间完成 Redis 原子自动收尾。 + 后续将 Reaper、配置停用与健康淘汰接入 `BeginDrain` 编排，形成可观测闭环。 ## 串并行关系