diff --git a/docs/adr/005-redis-activity-pool.md b/docs/adr/005-redis-activity-pool.md index a3e4197..6bb0b4a 100644 --- a/docs/adr/005-redis-activity-pool.md +++ b/docs/adr/005-redis-activity-pool.md @@ -100,6 +100,8 @@ pp:{activity}:carrier: ZSET 运营商候选索引 pp:{activity}:upstream: ZSET 供应商候选索引 pp:{activity}:owners HASH proxyID -> ownership assignment pp:{activity}:owner-expiry ZSET proxyID -> ownership expiry milliseconds +pp:{activity}:drain-tickets HASH proxyID -> pending drain ticket +pp:{activity}:worker-draining: ZSET 单 Worker 待绑定 Drain Ticket pp:{activity}:epoch STRING ownership 全局递增代次 pp:{activity}:inventory HASH upstreamID -> 当前未提取库存 pp:{activity}:worker-sessions HASH workerID -> 当前 Worker session @@ -180,6 +182,8 @@ Assign、Renew、BeginDrain 和 AcknowledgeDrain 分别使用有界小脚本， - Assign 只接受 AVAILABLE、无 owner 且未到 `usableUntil` 的 Proxy。 - Renew 必须匹配 worker、epoch，并把 lease 截断到 `usableUntil`。 - BeginDrain 对同一 assignment 幂等。 +- 首次 BeginDrain 原子写入按 Worker 索引的待绑定 Drain Ticket，并推进全局 ownership + epoch；`worker-owned` 已移除该 Proxy，下一份完整快照可据此撤销新预留。 - AcknowledgeDrain 仅在 Active 和 Reserved 都为零时释放所有权。 - Assign 与 Extract 并发竞争同一 Proxy 时，只允许一个操作成功。 - Expire 使用 `limit` 分批回收过期 assignment，禁止无界返回。 @@ -190,6 +194,10 @@ Gateway 的 `Capacity` 使用一次打包原子读取取得同一时刻的 Activ `snapshot.Store` 周期生成完整稀疏报告。当前 Snapshot 已移除但仍有活动连接的 Proxy 继续以 `draining=true` 上报，直到 Active/Reserved 同时归零。 +Drain Ticket 目前只记录“必须由不早于所需 epoch 的完整快照排除”的持久化前置条件。 +后续会在 Runtime 替换 Lua 事务中同时验证 Snapshot ACK、Ticket 屏障和零计数，避免 +拆分为读 Runtime 再释放所有权产生竞态。 + Redis Adapter 在单个 `{activity}` 原子边界内维护 Worker session 和运行态报告： 1. 新 Worker session 替换旧 session，并隔离旧实例后续写入。 diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index 9dda089..ef36602 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -135,6 +135,12 @@ Delta 声明 `base_version`。Worker 只有在本地版本恰好等于 base 且 3. 两个计数都归零后 Controller 清除所有权。 4. 无所有权 Proxy 才能进入 Distribution 的 Redis 原子提取操作。 +首次 `BeginDrain` 会原子创建按 `proxy_id + worker_id + assignment_epoch` 栅栏的待绑定 +Drain Ticket，并推进全局 ownership epoch，促使下一份完整 Snapshot 撤销该 Proxy。Ticket +按 Worker 有界读取，供 Controller 在确认完整 Snapshot 确实不含该 Proxy 后绑定快照屏障。 +当前实现尚未把该屏障和 Runtime 报告放入同一最终确认事务，因此不会自动释放所有权； +`AcknowledgeDrain` 仍是已有的显式完成原语。 + Worker 崩溃时必须等待所有权 epoch/有效期失效后再转移，避免双主。Proto 中 `ReportRuntimeResponse.revoke_proxy_ids` 是加速 Drain 的控制信号，不绕过 Snapshot 版本和 Redis 中的权威所有权状态。 diff --git a/findings.md b/findings.md index 2f805b3..f4564e7 100644 --- a/findings.md +++ b/findings.md @@ -127,6 +127,15 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两 的权威栅栏；epoch 变化不重置 version。此前 Gateway 错把 epoch 前进要求为 version=1， 与 Controller 的 `lastAppliedVersion+1` 生成规则冲突，现已用连续版本规则统一。 +## Drain Ticket（2026-08-02） + +- 原有 `BeginDrain` 只会把 Proxy 从 Worker 下发索引移除，无法让后续步骤区分“已发起 + 撤销”与“Gateway 已收到排除该 Proxy 的完整快照”。现已增加按 Worker 有界读取的持久化 + Ticket，并以 RequiredSnapshotEpoch 强制下一份权威快照至少跨过 Drain 操作。 +- Ticket 不能单独成为释放依据。最终清理必须把 Ticket 屏障、当前 session ACK 和完整 + Runtime 替换中的零计数置于同一 Redis Lua 原子边界；`ReadRuntime` 后再调用 + `AcknowledgeDrain` 会保留竞争窗口。 + ## Git 同步事实（2026-07-29） - PostgreSQL 管理面基础文档已提交为 `7951c29`。 diff --git a/internal/adapters/redisactivity/codec.go b/internal/adapters/redisactivity/codec.go index 875c97e..2a1b3f5 100644 --- a/internal/adapters/redisactivity/codec.go +++ b/internal/adapters/redisactivity/codec.go @@ -57,6 +57,15 @@ type ownershipRecord struct { Draining bool `json:"draining"` } +type drainTicketRecord struct { + Version int `json:"version"` + ProxyID string `json:"proxyId"` + WorkerID string `json:"workerId"` + WorkerIndexKey string `json:"workerIndexKey"` + AssignmentEpoch uint64 `json:"assignmentEpoch"` + RequiredSnapshotEpoch uint64 `json:"requiredSnapshotEpoch"` +} + type idempotencyRecord struct { Version int RequestDigest string @@ -240,6 +249,16 @@ func validateOwnershipRecord(record ownershipRecord) error { return nil } +func validateDrainTicketRecord(record drainTicketRecord) error { + if record.Version != recordVersion || record.ProxyID == "" || record.WorkerID == "" || + record.AssignmentEpoch == 0 || record.RequiredSnapshotEpoch == 0 || + record.RequiredSnapshotEpoch <= record.AssignmentEpoch || record.WorkerIndexKey == "" || + !strings.Contains(record.WorkerIndexKey, "{activity}") { + return ErrInvalidRecord + } + return nil +} + func validateIdempotencyRecord(record idempotencyRecord) error { if record.Version != recordVersion { return invalidRecord("unsupported idempotency version") diff --git a/internal/adapters/redisactivity/keys.go b/internal/adapters/redisactivity/keys.go index 54d4978..fc09f72 100644 --- a/internal/adapters/redisactivity/keys.go +++ b/internal/adapters/redisactivity/keys.go @@ -17,6 +17,7 @@ type keyspace struct { available string owners string ownerExpiry string + drainTickets string epoch string inventory string stateInventory string @@ -49,6 +50,7 @@ func newKeyspace(namespace string) keyspace { available: prefix + ":available", owners: prefix + ":owners", ownerExpiry: prefix + ":owner-expiry", + drainTickets: prefix + ":drain-tickets", epoch: prefix + ":epoch", inventory: prefix + ":inventory", stateInventory: prefix + ":state-inventory", @@ -122,6 +124,10 @@ func (keys keyspace) workerOwned(workerID string) string { return keys.facet("worker-owned", workerID) } +func (keys keyspace) workerDraining(workerID string) string { + return keys.facet("worker-draining", workerID) +} + func (keys keyspace) facet(name, value string) string { return keys.prefix + ":" + name + ":" + digestToken(value) } diff --git a/internal/adapters/redisactivity/ownership.go b/internal/adapters/redisactivity/ownership.go index 96e0166..0ff9eb8 100644 --- a/internal/adapters/redisactivity/ownership.go +++ b/internal/adapters/redisactivity/ownership.go @@ -19,6 +19,7 @@ const ( ) var _ ownershipDomain.Repository = (*Adapter)(nil) +var _ ownershipDomain.DrainTicketStore = (*Adapter)(nil) func (a *Adapter) Assign( ctx context.Context, @@ -109,6 +110,43 @@ func (a *Adapter) BeginDrain( } } +func (a *Adapter) PendingDrains(ctx context.Context, workerID string, limit int) ([]ownershipDomain.DrainTicket, error) { + if err := validateOwnershipCall(ctx, a); err != nil { + return nil, err + } + if workerID == "" || limit <= 0 { + return nil, ownershipDomain.ErrInvalidDrainTicket + } + result, err := runScript(ctx, a.client, drainTicketsScript, []string{ + a.keys.drainTickets, a.keys.workerDraining(workerID), a.keys.owners, + }, workerID, limit) + if err != nil { + return nil, err + } + var reply drainTicketsScriptReply + if err := decodeScriptResult(result, &reply); err != nil { + return nil, err + } + if reply.Status == scriptInvalid { + return nil, ownershipDomain.ErrInvalidDrainTicket + } + if reply.Status != scriptOK { + return nil, invalidScriptReply("unexpected pending drain tickets reply") + } + tickets := make([]ownershipDomain.DrainTicket, 0, len(reply.Tickets)) + for _, raw := range reply.Tickets { + var record drainTicketRecord + if err := decodeJSON(raw, &record); err != nil || validateDrainTicketRecord(record) != nil { + return nil, invalidScriptReply("pending drain tickets reply contained an invalid ticket") + } + tickets = append(tickets, ownershipDomain.DrainTicket{ + ProxyID: record.ProxyID, WorkerID: record.WorkerID, AssignmentEpoch: record.AssignmentEpoch, + RequiredSnapshotEpoch: record.RequiredSnapshotEpoch, + }) + } + return tickets, nil +} + func (a *Adapter) AcknowledgeDrain( ctx context.Context, proxyID string, @@ -226,8 +264,9 @@ func (a *Adapter) runOwnership( a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry, a.keys.epoch, operationKey, a.keys.healthDue, a.keys.healthQueued, a.keys.healthLeases, a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask, a.keys.healthUnhealthy, + a.keys.drainTickets, }, operation, operationTTLMillis(a.options.OperationTTL), a.options.CleanupLimit, - nowMS, proxyID, workerID, epoch, value, active, reserved, a.keys.workerOwned(workerID)) + nowMS, proxyID, workerID, epoch, value, active, reserved, a.keys.workerOwned(workerID), a.keys.workerDraining(workerID)) if err != nil { return ownershipScriptReply{}, err } diff --git a/internal/adapters/redisactivity/ownership_integration_test.go b/internal/adapters/redisactivity/ownership_integration_test.go index 26d6745..37f4e48 100644 --- a/internal/adapters/redisactivity/ownership_integration_test.go +++ b/internal/adapters/redisactivity/ownership_integration_test.go @@ -64,6 +64,14 @@ func TestRedisOwnershipLifecycle(t *testing.T) { t.Fatalf("BeginDrain() = %+v, %v", draining, err) } assertRedisWorkerOwned(t, fixture, "worker-a") + pending, err := fixture.Adapter.PendingDrains(context.Background(), "worker-a", 1) + if err != nil || len(pending) != 1 || pending[0].ProxyID != "proxy-a" || + pending[0].AssignmentEpoch != assigned.Epoch || pending[0].RequiredSnapshotEpoch <= assigned.Epoch { + t.Fatalf("PendingDrains() = %+v, %v", pending, err) + } + if tickets, err := fixture.Client.ZRange(context.Background(), fixture.Adapter.keys.workerDraining("worker-a"), 0, -1).Result(); err != nil || len(tickets) != 1 || tickets[0] != "proxy-a" { + t.Fatalf("worker-draining = %v, %v", tickets, err) + } replayed, err := fixture.Adapter.BeginDrain(context.Background(), "proxy-a", "worker-a", assigned.Epoch) if err != nil || replayed != draining { t.Fatalf("BeginDrain(replay) = %+v, %v", replayed, err) @@ -88,6 +96,9 @@ func TestRedisOwnershipLifecycle(t *testing.T) { if current, ok, err := fixture.Adapter.Get(context.Background(), "proxy-a"); err != nil || ok { t.Fatalf("Get(after ACK) = %+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 ACK) = %+v, %v", pending, err) + } restored, err := fixture.Adapter.Extract(context.Background(), extractionDomain.Command{ RequestID: "req-restored", ClientID: "client-a", Requested: 1, Fulfillment: extractionDomain.Partial, Now: now.Add(40 * time.Second), diff --git a/internal/adapters/redisactivity/scripts.go b/internal/adapters/redisactivity/scripts.go index aabd08a..79a4e18 100644 --- a/internal/adapters/redisactivity/scripts.go +++ b/internal/adapters/redisactivity/scripts.go @@ -87,6 +87,11 @@ type ownershipScriptReply struct { Record string `json:"record,omitempty"` } +type drainTicketsScriptReply struct { + Status scriptStatus `json:"status"` + Tickets []string `json:"tickets"` +} + type maintenanceScriptReply struct { Status scriptStatus `json:"status"` Count int `json:"count"` @@ -153,6 +158,9 @@ var extractSource string //go:embed scripts/ownership.lua var ownershipSource string +//go:embed scripts/drain_tickets.lua +var drainTicketsSource string + //go:embed scripts/sweep.lua var sweepSource string @@ -176,6 +184,7 @@ var ( upstreamLookupScript = redis.NewScript(upstreamLookupSource) extractScript = redis.NewScript(extractSource) ownershipScript = redis.NewScript(ownershipSource) + drainTicketsScript = redis.NewScript(drainTicketsSource) sweepScript = redis.NewScript(sweepSource) statusScript = redis.NewScript(statusSource) runtimeScript = redis.NewScript(runtimeSource) diff --git a/internal/adapters/redisactivity/scripts/drain_tickets.lua b/internal/adapters/redisactivity/scripts/drain_tickets.lua new file mode 100644 index 0000000..b7fdbf4 --- /dev/null +++ b/internal/adapters/redisactivity/scripts/drain_tickets.lua @@ -0,0 +1,55 @@ +local tickets_key = KEYS[1] +local worker_index_key = KEYS[2] +local owners_key = KEYS[3] + +local worker_id = ARGV[1] +local limit = tonumber(ARGV[2]) + +local function reply(status, tickets) + return cjson.encode({status = status, tickets = tickets or {}}) +end + +local function decode_table(raw) + if not raw then + return nil + end + local ok, value = pcall(cjson.decode, raw) + if not ok or type(value) ~= 'table' then + return nil + end + return value +end + +local function valid_ticket(ticket, proxy_id) + return ticket and ticket.version == 1 and ticket.proxyId == proxy_id and + ticket.workerId == worker_id and type(ticket.workerIndexKey) == 'string' and + ticket.workerIndexKey == worker_index_key and tonumber(ticket.assignmentEpoch) and + tonumber(ticket.assignmentEpoch) > 0 and tonumber(ticket.requiredSnapshotEpoch) and + tonumber(ticket.requiredSnapshotEpoch) > tonumber(ticket.assignmentEpoch) +end + +local function valid_assignment(assignment, ticket) + return assignment and assignment.version == 1 and assignment.draining == true and + assignment.proxyId == ticket.proxyId and assignment.workerId == ticket.workerId and + tonumber(assignment.epoch) == tonumber(ticket.assignmentEpoch) +end + +if type(worker_id) ~= 'string' or worker_id == '' or not limit or limit <= 0 then + return reply('invalid') +end + +local scan_limit = limit * 4 +local ids = redis.call('ZRANGE', worker_index_key, 0, scan_limit - 1) +local tickets = {} +for _, proxy_id in ipairs(ids) do + local raw = redis.call('HGET', tickets_key, proxy_id) + local ticket = decode_table(raw) + local assignment = decode_table(redis.call('HGET', owners_key, proxy_id)) + if valid_ticket(ticket, proxy_id) and valid_assignment(assignment, ticket) and #tickets < limit then + tickets[#tickets + 1] = raw + else + redis.call('HDEL', tickets_key, proxy_id) + redis.call('ZREM', worker_index_key, proxy_id) + end +end +return reply('ok', tickets) diff --git a/internal/adapters/redisactivity/scripts/ownership.lua b/internal/adapters/redisactivity/scripts/ownership.lua index ebf3487..8391e99 100644 --- a/internal/adapters/redisactivity/scripts/ownership.lua +++ b/internal/adapters/redisactivity/scripts/ownership.lua @@ -16,6 +16,7 @@ local health_tasks_key = KEYS[15] local health_task_expiry_key = KEYS[16] local health_ref_task_key = KEYS[17] local health_unhealthy_key = KEYS[18] +local drain_tickets_key = KEYS[19] local operation = ARGV[1] local operation_ttl_ms = tonumber(ARGV[2]) @@ -28,6 +29,7 @@ local value = tonumber(ARGV[8]) local active = tonumber(ARGV[9]) local reserved = tonumber(ARGV[10]) local worker_index_key = ARGV[11] +local worker_draining_index_key = ARGV[12] local mutating = operation ~= 'get' local function finish(reply) @@ -108,6 +110,17 @@ local function remove_owned(id, record) end end +local function clear_drain(id) + local raw = redis.call('HGET', drain_tickets_key, id) + if raw then + local decoded, ticket = pcall(cjson.decode, raw) + if decoded and type(ticket) == 'table' and type(ticket.workerIndexKey) == 'string' and ticket.workerIndexKey ~= '' then + redis.call('ZREM', ticket.workerIndexKey, id) + end + end + redis.call('HDEL', drain_tickets_key, id) +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 @@ -176,6 +189,7 @@ local function remove_proxy(id) redis.call('ZREM', expiry_key, id) redis.call('HDEL', owners_key, id) redis.call('ZREM', owner_expiry_key, id) + clear_drain(id) redis.call('ZREM', health_unhealthy_key, id) remove_health_task(id) end @@ -220,6 +234,7 @@ local function clear_owner(id, assignment, at_ms, restore) end redis.call('HDEL', owners_key, id) redis.call('ZREM', owner_expiry_key, id) + clear_drain(id) if assignment and type(assignment.workerIndexKey) == 'string' and assignment.workerIndexKey ~= '' then redis.call('ZREM', assignment.workerIndexKey, id) end @@ -322,8 +337,14 @@ if operation == 'begin_drain' then return finish({status = 'stale'}) end if not current.draining then + if type(worker_draining_index_key) ~= 'string' or worker_draining_index_key == '' or + not string.find(worker_draining_index_key, '{activity}', 1, true) then + return finish({status = 'invalid'}) + end current.draining = true current.assignmentVersion = tonumber(current.assignmentVersion) + 1 + local drain_epoch = redis.call('INCR', epoch_key) + redis.call('PERSIST', epoch_key) local encoded = cjson.encode(current) redis.call('HSET', owners_key, proxy_id, encoded) local record = decode_table(redis.call('HGET', records_key, proxy_id)) @@ -331,6 +352,18 @@ if operation == 'begin_drain' then if type(current.workerIndexKey) == 'string' and current.workerIndexKey ~= '' then redis.call('ZREM', current.workerIndexKey, proxy_id) end + local ticket = { + version = 1, + proxyId = proxy_id, + workerId = worker_id, + workerIndexKey = worker_draining_index_key, + assignmentEpoch = current.epoch, + requiredSnapshotEpoch = drain_epoch, + } + redis.call('HSET', drain_tickets_key, proxy_id, cjson.encode(ticket)) + redis.call('ZADD', worker_draining_index_key, current.expiresAtMs, proxy_id) + touch(drain_tickets_key, tonumber(current.expiresAtMs)) + touch(worker_draining_index_key, tonumber(current.expiresAtMs)) return finish({status = 'ok', record = encoded}) end return finish({status = 'ok', record = cjson.encode(current)}) diff --git a/internal/domain/activitypool/contracttest/contract.go b/internal/domain/activitypool/contracttest/contract.go index 2951c72..e016a4d 100644 --- a/internal/domain/activitypool/contracttest/contract.go +++ b/internal/domain/activitypool/contracttest/contract.go @@ -22,6 +22,7 @@ type Store interface { activitypool.Maintainer extractionDomain.Store ownershipDomain.Repository + ownershipDomain.DrainTicketStore } type Factory func(*testing.T) (Store, func()) @@ -293,6 +294,11 @@ func runOwnershipContract(t *testing.T, factory Factory) { if replayed, err := store.BeginDrain(context.Background(), "proxy-a", "worker-a", assigned.Epoch); err != nil || replayed != draining { t.Fatalf("BeginDrain(replay) = %+v, %v", replayed, err) } + pending, err := store.PendingDrains(context.Background(), "worker-a", 1) + if err != nil || len(pending) != 1 || pending[0].ProxyID != "proxy-a" || + pending[0].AssignmentEpoch != assigned.Epoch || pending[0].RequiredSnapshotEpoch <= assigned.Epoch { + t.Fatalf("PendingDrains() = %+v, %v", pending, err) + } if err := store.AcknowledgeDrain(context.Background(), "proxy-a", "worker-a", assigned.Epoch, 1, 0); !errors.Is(err, ownershipDomain.ErrDrainNotReady) { t.Fatalf("AcknowledgeDrain(active) error = %v", err) } @@ -302,6 +308,9 @@ func runOwnershipContract(t *testing.T, factory Factory) { if current, ok, err := store.Get(context.Background(), "proxy-a"); err != nil || ok { t.Fatalf("Get(after ACK) = %+v, %t, %v", current, ok, err) } + if pending, err := store.PendingDrains(context.Background(), "worker-a", 1); err != nil || len(pending) != 0 { + t.Fatalf("PendingDrains(after ACK) = %+v, %v", pending, err) + } if result, err := store.Extract(context.Background(), extractionDomain.Command{ RequestID: "req-after-ack", ClientID: "client-a", Requested: 1, Fulfillment: extractionDomain.Partial, Now: now.Add(20 * time.Second), @@ -537,6 +546,7 @@ func runCancellationContract(t *testing.T, store Store) { {name: "assign", call: func() error { _, err := store.Assign(ctx, now, "proxy-a", "worker-a", time.Minute); return err }}, {name: "renew", call: func() error { _, err := store.Renew(ctx, now, "proxy-a", "worker-a", 1, time.Minute); return err }}, {name: "begin drain", call: func() error { _, err := store.BeginDrain(ctx, "proxy-a", "worker-a", 1); return err }}, + {name: "pending drains", call: func() error { _, err := store.PendingDrains(ctx, "worker-a", 1); return err }}, {name: "acknowledge drain", call: func() error { return store.AcknowledgeDrain(ctx, "proxy-a", "worker-a", 1, 0, 0) }}, {name: "get", call: func() error { _, _, err := store.Get(ctx, "proxy-a"); return err }}, {name: "expire", call: func() error { _, err := store.Expire(ctx, now, 1); return err }}, diff --git a/internal/domain/activitypool/pool.go b/internal/domain/activitypool/pool.go index 832fcea..6daa87c 100644 --- a/internal/domain/activitypool/pool.go +++ b/internal/domain/activitypool/pool.go @@ -159,6 +159,7 @@ type MemoryPool struct { keyByID map[string]string idempotent map[string]idempotencyEntry ownership map[string]ownershipDomain.Assignment + drains map[string]ownershipDomain.DrainTicket targets map[targetHealthKey]healthDomain.TargetState unhealthy map[string]time.Time nextEpoch uint64 @@ -176,17 +177,18 @@ type idempotencyEntry struct { } var ( - _ Upserter = (*MemoryPool)(nil) - _ HealthStore = (*MemoryPool)(nil) - _ GlobalHealthStore = (*MemoryPool)(nil) - _ TargetHealthStore = (*MemoryPool)(nil) - _ ProxyUpstreamReader = (*MemoryPool)(nil) - _ InventoryReader = (*MemoryPool)(nil) - _ StateInventoryReader = (*MemoryPool)(nil) - _ Maintainer = (*MemoryPool)(nil) - _ UnhealthyRemover = (*MemoryPool)(nil) - _ extractionDomain.Store = (*MemoryPool)(nil) - _ ownershipDomain.Repository = (*MemoryPool)(nil) + _ Upserter = (*MemoryPool)(nil) + _ HealthStore = (*MemoryPool)(nil) + _ GlobalHealthStore = (*MemoryPool)(nil) + _ TargetHealthStore = (*MemoryPool)(nil) + _ ProxyUpstreamReader = (*MemoryPool)(nil) + _ InventoryReader = (*MemoryPool)(nil) + _ StateInventoryReader = (*MemoryPool)(nil) + _ Maintainer = (*MemoryPool)(nil) + _ UnhealthyRemover = (*MemoryPool)(nil) + _ extractionDomain.Store = (*MemoryPool)(nil) + _ ownershipDomain.Repository = (*MemoryPool)(nil) + _ ownershipDomain.DrainTicketStore = (*MemoryPool)(nil) ) func NewMemoryPool() *MemoryPool { @@ -195,6 +197,7 @@ func NewMemoryPool() *MemoryPool { keyByID: make(map[string]string), idempotent: make(map[string]idempotencyEntry), ownership: make(map[string]ownershipDomain.Assignment), + drains: make(map[string]ownershipDomain.DrainTicket), targets: make(map[targetHealthKey]healthDomain.TargetState), unhealthy: make(map[string]time.Time), } @@ -819,6 +822,7 @@ func (p *MemoryPool) Assign(ctx context.Context, now time.Time, proxyID, workerI p.setEntryByIDLocked(proxyID, entry) } delete(p.ownership, proxyID) + delete(p.drains, proxyID) } entry, ok := p.entryByIDLocked(proxyID) if !ok || entry.State != proxyDomain.StateAvailable || entry.OwnerWorkerID != "" || !entry.UsableUntil.After(now) { @@ -882,10 +886,49 @@ func (p *MemoryPool) BeginDrain(ctx context.Context, proxyID, workerID string, e assignment.Draining = true assignment.Version++ p.ownership[proxyID] = assignment + p.nextEpoch++ + p.drains[proxyID] = ownershipDomain.DrainTicket{ + ProxyID: proxyID, WorkerID: workerID, AssignmentEpoch: assignment.Epoch, + RequiredSnapshotEpoch: p.nextEpoch, + } } return assignment, nil } +func (p *MemoryPool) PendingDrains(ctx context.Context, workerID string, limit int) ([]ownershipDomain.DrainTicket, error) { + if err := ownershipContextError(ctx); err != nil { + return nil, err + } + if p == nil || workerID == "" || limit <= 0 { + return nil, ownershipDomain.ErrInvalidDrainTicket + } + p.mu.Lock() + defer p.mu.Unlock() + if err := ctx.Err(); err != nil { + return nil, err + } + proxyIDs := make([]string, 0) + for proxyID, ticket := range p.drains { + assignment, exists := p.ownership[proxyID] + if !exists || !assignment.Draining || assignment.WorkerID != ticket.WorkerID || assignment.Epoch != ticket.AssignmentEpoch { + delete(p.drains, proxyID) + continue + } + if ticket.WorkerID == workerID { + proxyIDs = append(proxyIDs, proxyID) + } + } + sort.Strings(proxyIDs) + if len(proxyIDs) > limit { + proxyIDs = proxyIDs[:limit] + } + tickets := make([]ownershipDomain.DrainTicket, 0, len(proxyIDs)) + for _, proxyID := range proxyIDs { + tickets = append(tickets, p.drains[proxyID]) + } + return tickets, nil +} + func (p *MemoryPool) AcknowledgeDrain(ctx context.Context, proxyID, workerID string, epoch uint64, active, reserved int64) error { if err := ownershipContextError(ctx); err != nil { return err @@ -913,6 +956,7 @@ func (p *MemoryPool) AcknowledgeDrain(ctx context.Context, proxyID, workerID str p.setEntryByIDLocked(proxyID, entry) } delete(p.ownership, proxyID) + delete(p.drains, proxyID) return nil } @@ -969,6 +1013,7 @@ func (p *MemoryPool) Expire(ctx context.Context, now time.Time, limit int) ([]ow } expired = append(expired, assignment) delete(p.ownership, proxyID) + delete(p.drains, proxyID) } return expired, nil } @@ -1012,6 +1057,7 @@ func (p *MemoryPool) removeEntryByIDLocked(proxyID string) { func (p *MemoryPool) removeEntryLocked(key string, entry Entry) { delete(p.ownership, entry.Proxy.ID) + delete(p.drains, entry.Proxy.ID) delete(p.unhealthy, entry.Proxy.ID) for target := range p.targets { if target.proxyID == entry.Proxy.ID { diff --git a/internal/domain/ownership/ownership.go b/internal/domain/ownership/ownership.go index c62a0c6..849840a 100644 --- a/internal/domain/ownership/ownership.go +++ b/internal/domain/ownership/ownership.go @@ -13,6 +13,7 @@ var ( ErrStaleAssignment = errors.New("proxy ownership assignment is stale") ErrNotDraining = errors.New("proxy ownership is not draining") ErrDrainNotReady = errors.New("proxy still has active or reserved runtime") + ErrInvalidDrainTicket = errors.New("invalid drain ticket request") ) type Assignment struct { @@ -24,6 +25,23 @@ type Assignment struct { Draining bool } +// DrainTicket records a pending removal after an ownership assignment enters +// draining. RequiredSnapshotEpoch fences the first complete Worker snapshot +// that may prove the proxy was excluded. +type DrainTicket struct { + ProxyID string + WorkerID string + AssignmentEpoch uint64 + RequiredSnapshotEpoch uint64 +} + +// DrainTicketStore exposes bounded, Worker-local pending drain tickets. It +// deliberately does not finalize ownership: a later runtime transaction must +// prove both an acknowledged exclusion snapshot and zero live counters. +type DrainTicketStore interface { + PendingDrains(context.Context, string, int) ([]DrainTicket, error) +} + // Repository is the shared authority for ownership changes. Implementations // that also support extraction must serialize both operations transactionally. type Repository interface { diff --git a/progress.md b/progress.md index 8c9a785..a4e5a46 100644 --- a/progress.md +++ b/progress.md @@ -24,6 +24,10 @@ - 修复 Snapshot epoch/version 栅栏：Worker 流的 `version` 在 ownership epoch 前进时 仍严格递增，Gateway 以连续版本加非倒退 epoch 进行完整快照原子替换，避免动态分配后的 有效刷新被错误拒绝。这是后续自动 Drain 编排的必要前提。 +- Drain 首次发起现在原子推进 ownership epoch，并写入由 Proxy、Worker、原 assignment + epoch 与所需 Snapshot epoch 组成的待绑定 Ticket；Redis 使用 `drain-tickets` 与按 + Worker 的 `worker-draining` 索引，ACK、过期和硬删除会清理 Ticket。完整快照排除、 + ACK 与 Runtime 零计数的原子自动完成仍在后续切片，尚未标记为完成。 ## 2026-07-30 diff --git a/task_plan.md b/task_plan.md index 8121e36..0cd5181 100644 --- a/task_plan.md +++ b/task_plan.md @@ -41,7 +41,9 @@ 通过 Controller 后台的有界回收清理无 Worker ownership 的持续异常 Proxy；拥有 Worker 的候选延后到既有 Drain/ACK 清除所有权后再处理。 15. [进行中] 收敛 Worker 发布生命周期：已完成权威 Proxy/Routing 的持续完整快照 - 刷新；后续补齐自动 Drain 编排，使配置停用、健康淘汰与 Snapshot 撤销形成可观测闭环。 + 刷新，以及 Drain 发起时的持久化 Ticket、Worker 待绑定索引和 ownership epoch 推进； + 后续在完整快照排除、ACK 与 Runtime 零计数之间补齐原子自动完成编排，使配置停用、 + 健康淘汰与 Snapshot 撤销形成可观测闭环。 ## 串并行关系