From 88d5ac24d46ef4725a581e8a28e1f54f0b4def4f Mon Sep 17 00:00:00 2001 From: youfak Date: Sun, 2 Aug 2026 11:02:19 +0800 Subject: [PATCH] feat: bind drain tickets to full snapshots --- docs/adr/005-redis-activity-pool.md | 3 +- docs/api/control-plane.md | 5 +- findings.md | 3 +- internal/adapters/redisactivity/codec.go | 24 +++++-- internal/adapters/redisactivity/ownership.go | 59 ++++++++++++++++- .../ownership_integration_test.go | 10 +++ internal/adapters/redisactivity/scripts.go | 30 +++++---- .../scripts/bind_drain_ticket.lua | 65 +++++++++++++++++++ internal/controller/bootstrap/bootstrap.go | 5 ++ internal/controller/worker/grpc_handler.go | 26 +++++++- .../controller/worker/grpc_handler_test.go | 36 ++++++++++ internal/controller/worker/service.go | 52 ++++++++++++++- .../activitypool/contracttest/contract.go | 11 ++++ internal/domain/activitypool/pool.go | 50 ++++++++++++++ internal/domain/ownership/ownership.go | 10 +++ progress.md | 5 +- task_plan.md | 4 +- 17 files changed, 366 insertions(+), 32 deletions(-) create mode 100644 internal/adapters/redisactivity/scripts/bind_drain_ticket.lua diff --git a/docs/adr/005-redis-activity-pool.md b/docs/adr/005-redis-activity-pool.md index 6bb0b4a..3d57244 100644 --- a/docs/adr/005-redis-activity-pool.md +++ b/docs/adr/005-redis-activity-pool.md @@ -194,7 +194,8 @@ Gateway 的 `Capacity` 使用一次打包原子读取取得同一时刻的 Activ `snapshot.Store` 周期生成完整稀疏报告。当前 Snapshot 已移除但仍有活动连接的 Proxy 继续以 `draining=true` 上报，直到 Active/Reserved 同时归零。 -Drain Ticket 目前只记录“必须由不早于所需 epoch 的完整快照排除”的持久化前置条件。 +Drain Ticket 会绑定“已签发给当前 session、且不早于所需 epoch 的完整排除快照”引用， +包括 version、epoch 和 checksum。 后续会在 Runtime 替换 Lua 事务中同时验证 Snapshot ACK、Ticket 屏障和零计数，避免 拆分为读 Runtime 再释放所有权产生竞态。 diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index ef36602..25775b4 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -138,8 +138,9 @@ Delta 声明 `base_version`。Worker 只有在本地版本恰好等于 base 且 首次 `BeginDrain` 会原子创建按 `proxy_id + worker_id + assignment_epoch` 栅栏的待绑定 Drain Ticket，并推进全局 ownership epoch，促使下一份完整 Snapshot 撤销该 Proxy。Ticket 按 Worker 有界读取，供 Controller 在确认完整 Snapshot 确实不含该 Proxy 后绑定快照屏障。 -当前实现尚未把该屏障和 Runtime 报告放入同一最终确认事务，因此不会自动释放所有权； -`AcknowledgeDrain` 仍是已有的显式完成原语。 +Worker Handler 先登记该 Snapshot 引用，再将 Ticket 绑定到 session、version、epoch 与 +checksum。当前实现尚未把该屏障和 Runtime 报告放入同一最终确认事务，因此不会自动释放 +所有权；`AcknowledgeDrain` 仍是已有的显式完成原语。 Worker 崩溃时必须等待所有权 epoch/有效期失效后再转移，避免双主。Proto 中 `ReportRuntimeResponse.revoke_proxy_ids` 是加速 Drain 的控制信号，不绕过 diff --git a/findings.md b/findings.md index f4564e7..45c60c1 100644 --- a/findings.md +++ b/findings.md @@ -131,7 +131,8 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两 - 原有 `BeginDrain` 只会把 Proxy 从 Worker 下发索引移除，无法让后续步骤区分“已发起 撤销”与“Gateway 已收到排除该 Proxy 的完整快照”。现已增加按 Worker 有界读取的持久化 - Ticket，并以 RequiredSnapshotEpoch 强制下一份权威快照至少跨过 Drain 操作。 + Ticket，并以 RequiredSnapshotEpoch 强制下一份权威快照至少跨过 Drain 操作。Handler 在 + Snapshot 引用登记成功后才绑定屏障，绑定内容包含 session、version、epoch 与 checksum。 - Ticket 不能单独成为释放依据。最终清理必须把 Ticket 屏障、当前 session ACK 和完整 Runtime 替换中的零计数置于同一 Redis Lua 原子边界；`ReadRuntime` 后再调用 `AcknowledgeDrain` 会保留竞争窗口。 diff --git a/internal/adapters/redisactivity/codec.go b/internal/adapters/redisactivity/codec.go index 2a1b3f5..0a867bd 100644 --- a/internal/adapters/redisactivity/codec.go +++ b/internal/adapters/redisactivity/codec.go @@ -58,12 +58,16 @@ type ownershipRecord struct { } 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"` + Version int `json:"version"` + ProxyID string `json:"proxyId"` + WorkerID string `json:"workerId"` + WorkerIndexKey string `json:"workerIndexKey"` + AssignmentEpoch uint64 `json:"assignmentEpoch"` + RequiredSnapshotEpoch uint64 `json:"requiredSnapshotEpoch"` + SessionID string `json:"sessionId,omitempty"` + SnapshotVersion uint64 `json:"snapshotVersion,omitempty"` + SnapshotOwnershipEpoch uint64 `json:"snapshotOwnershipEpoch,omitempty"` + SnapshotChecksum string `json:"snapshotChecksum,omitempty"` } type idempotencyRecord struct { @@ -256,6 +260,14 @@ func validateDrainTicketRecord(record drainTicketRecord) error { !strings.Contains(record.WorkerIndexKey, "{activity}") { return ErrInvalidRecord } + bound := record.SnapshotVersion != 0 || record.SnapshotOwnershipEpoch != 0 || record.SessionID != "" || record.SnapshotChecksum != "" + if !bound { + return nil + } + if record.SessionID == "" || record.SnapshotVersion == 0 || record.SnapshotOwnershipEpoch < record.RequiredSnapshotEpoch || + !validDigest(record.SnapshotChecksum) { + return ErrInvalidRecord + } return nil } diff --git a/internal/adapters/redisactivity/ownership.go b/internal/adapters/redisactivity/ownership.go index 0ff9eb8..4abf685 100644 --- a/internal/adapters/redisactivity/ownership.go +++ b/internal/adapters/redisactivity/ownership.go @@ -2,6 +2,8 @@ package redisactivity import ( "context" + "crypto/sha256" + "encoding/hex" "errors" "fmt" "time" @@ -139,14 +141,57 @@ func (a *Adapter) PendingDrains(ctx context.Context, workerID string, limit int) 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{ + ticket := ownershipDomain.DrainTicket{ ProxyID: record.ProxyID, WorkerID: record.WorkerID, AssignmentEpoch: record.AssignmentEpoch, RequiredSnapshotEpoch: record.RequiredSnapshotEpoch, - }) + } + if record.SnapshotVersion != 0 { + checksum, err := hex.DecodeString(record.SnapshotChecksum) + if err != nil || len(checksum) != sha256.Size { + return nil, invalidScriptReply("pending drain ticket barrier checksum is invalid") + } + copy(ticket.Barrier.Checksum[:], checksum) + ticket.Barrier = ownershipDomain.SnapshotBarrier{ + SessionID: record.SessionID, Version: record.SnapshotVersion, + OwnershipEpoch: record.SnapshotOwnershipEpoch, Checksum: ticket.Barrier.Checksum, + } + } + tickets = append(tickets, ticket) } return tickets, nil } +func (a *Adapter) BindDrainBarrier(ctx context.Context, ticket ownershipDomain.DrainTicket) error { + if err := validateOwnershipCall(ctx, a); err != nil { + return err + } + if !validDrainTicket(ticket) || !validDrainBarrier(ticket.Barrier) || + ticket.Barrier.OwnershipEpoch < ticket.RequiredSnapshotEpoch { + return ownershipDomain.ErrInvalidDrainTicket + } + result, err := runScript(ctx, a.client, bindDrainTicketScript, []string{ + a.keys.drainTickets, a.keys.owners, a.keys.workerDraining(ticket.WorkerID), + }, ticket.ProxyID, ticket.WorkerID, ticket.AssignmentEpoch, ticket.RequiredSnapshotEpoch, + ticket.Barrier.SessionID, ticket.Barrier.Version, ticket.Barrier.OwnershipEpoch, hex.EncodeToString(ticket.Barrier.Checksum[:])) + if err != nil { + return err + } + var reply ownershipScriptReply + if err := decodeScriptResult(result, &reply); err != nil { + return err + } + switch reply.Status { + case scriptOK: + return nil + case scriptStale, scriptNotFound: + return ownershipDomain.ErrStaleAssignment + case scriptInvalid: + return ownershipDomain.ErrInvalidDrainTicket + default: + return invalidScriptReply("unexpected drain ticket barrier reply") + } +} + func (a *Adapter) AcknowledgeDrain( ctx context.Context, proxyID string, @@ -308,3 +353,13 @@ func validateOwnershipCall(ctx context.Context, adapter *Adapter) error { } return nil } + +func validDrainTicket(ticket ownershipDomain.DrainTicket) bool { + return ticket.ProxyID != "" && ticket.WorkerID != "" && ticket.AssignmentEpoch > 0 && + ticket.RequiredSnapshotEpoch > ticket.AssignmentEpoch +} + +func validDrainBarrier(barrier ownershipDomain.SnapshotBarrier) bool { + return barrier.SessionID != "" && barrier.Version > 0 && barrier.OwnershipEpoch > 0 && + barrier.Checksum != [32]byte{} +} diff --git a/internal/adapters/redisactivity/ownership_integration_test.go b/internal/adapters/redisactivity/ownership_integration_test.go index 37f4e48..1dc8fe5 100644 --- a/internal/adapters/redisactivity/ownership_integration_test.go +++ b/internal/adapters/redisactivity/ownership_integration_test.go @@ -72,6 +72,16 @@ func TestRedisOwnershipLifecycle(t *testing.T) { 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) } + pending[0].Barrier = ownershipDomain.SnapshotBarrier{ + SessionID: "session-a", Version: 1, OwnershipEpoch: pending[0].RequiredSnapshotEpoch, + Checksum: [32]byte{1}, + } + if err := fixture.Adapter.BindDrainBarrier(context.Background(), pending[0]); err != nil { + t.Fatalf("BindDrainBarrier(): %v", err) + } + if bound, err := fixture.Adapter.PendingDrains(context.Background(), "worker-a", 1); err != nil || len(bound) != 1 || bound[0].Barrier != pending[0].Barrier { + t.Fatalf("PendingDrains(bound) = %+v, %v", bound, 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) diff --git a/internal/adapters/redisactivity/scripts.go b/internal/adapters/redisactivity/scripts.go index 79a4e18..cdb7990 100644 --- a/internal/adapters/redisactivity/scripts.go +++ b/internal/adapters/redisactivity/scripts.go @@ -161,6 +161,9 @@ var ownershipSource string //go:embed scripts/drain_tickets.lua var drainTicketsSource string +//go:embed scripts/bind_drain_ticket.lua +var bindDrainTicketSource string + //go:embed scripts/sweep.lua var sweepSource string @@ -177,19 +180,20 @@ var capacitySource string var workerSnapshotSource string var ( - upsertScript = redis.NewScript(upsertSource) - healthScript = redis.NewScript(healthSource) - healthTasksScript = redis.NewScript(healthTasksSource) - targetHealthScript = redis.NewScript(targetHealthSource) - 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) - capacityScript = redis.NewScript(capacitySource) - workerSnapshotScript = redis.NewScript(workerSnapshotSource) + upsertScript = redis.NewScript(upsertSource) + healthScript = redis.NewScript(healthSource) + healthTasksScript = redis.NewScript(healthTasksSource) + targetHealthScript = redis.NewScript(targetHealthSource) + upstreamLookupScript = redis.NewScript(upstreamLookupSource) + extractScript = redis.NewScript(extractSource) + ownershipScript = redis.NewScript(ownershipSource) + drainTicketsScript = redis.NewScript(drainTicketsSource) + bindDrainTicketScript = redis.NewScript(bindDrainTicketSource) + sweepScript = redis.NewScript(sweepSource) + statusScript = redis.NewScript(statusSource) + runtimeScript = redis.NewScript(runtimeSource) + capacityScript = redis.NewScript(capacitySource) + workerSnapshotScript = redis.NewScript(workerSnapshotSource) ) func runScript(ctx context.Context, client redis.Scripter, script *redis.Script, keys []string, args ...any) (any, error) { diff --git a/internal/adapters/redisactivity/scripts/bind_drain_ticket.lua b/internal/adapters/redisactivity/scripts/bind_drain_ticket.lua new file mode 100644 index 0000000..b0b6906 --- /dev/null +++ b/internal/adapters/redisactivity/scripts/bind_drain_ticket.lua @@ -0,0 +1,65 @@ +local tickets_key = KEYS[1] +local owners_key = KEYS[2] +local worker_index_key = KEYS[3] + +local proxy_id = ARGV[1] +local worker_id = ARGV[2] +local assignment_epoch = tonumber(ARGV[3]) +local required_epoch = tonumber(ARGV[4]) +local session_id = ARGV[5] +local version = tonumber(ARGV[6]) +local snapshot_epoch = tonumber(ARGV[7]) +local checksum = ARGV[8] + +local function reply(status) + return cjson.encode({status = status}) +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_checksum(value) + return type(value) == 'string' and string.len(value) == 64 and string.match(value, '^[0-9a-f]+$') ~= nil +end + +if type(proxy_id) ~= 'string' or proxy_id == '' or type(worker_id) ~= 'string' or worker_id == '' or + not assignment_epoch or assignment_epoch <= 0 or not required_epoch or required_epoch <= assignment_epoch or + type(session_id) ~= 'string' or session_id == '' or not version or version <= 0 or + not snapshot_epoch or snapshot_epoch < required_epoch or not valid_checksum(checksum) then + return reply('invalid') +end + +local ticket = decode_table(redis.call('HGET', tickets_key, proxy_id)) +local assignment = decode_table(redis.call('HGET', owners_key, proxy_id)) +if not ticket or ticket.version ~= 1 or ticket.proxyId ~= proxy_id or ticket.workerId ~= worker_id or + ticket.workerIndexKey ~= worker_index_key or tonumber(ticket.assignmentEpoch) ~= assignment_epoch or + tonumber(ticket.requiredSnapshotEpoch) ~= required_epoch or not assignment or assignment.version ~= 1 or + assignment.draining ~= true or assignment.workerId ~= worker_id or tonumber(assignment.epoch) ~= assignment_epoch then + return reply('stale') +end + +local current_epoch = tonumber(ticket.snapshotOwnershipEpoch) or 0 +local current_version = tonumber(ticket.snapshotVersion) or 0 +if current_epoch > snapshot_epoch or (current_epoch == snapshot_epoch and current_version > version) then + return reply('ok') +end +if current_epoch == snapshot_epoch and current_version == version then + if ticket.sessionId ~= session_id or ticket.snapshotChecksum ~= checksum then + return reply('stale') + end + return reply('ok') +end +ticket.sessionId = session_id +ticket.snapshotVersion = version +ticket.snapshotOwnershipEpoch = snapshot_epoch +ticket.snapshotChecksum = checksum +redis.call('HSET', tickets_key, proxy_id, cjson.encode(ticket)) +return reply('ok') diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index d56da3e..d486144 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -269,6 +269,10 @@ func runWithWorkerFactory( if reader, ok := opened.workerStore.(ownershipDomain.SnapshotReader); ok { snapshotReader = reader } + var drainTickets ownershipDomain.DrainTicketStore + if tickets, ok := opened.workerStore.(ownershipDomain.DrainTicketStore); ok { + drainTickets = tickets + } service, serviceErr := worker.NewService(opened.workerStore, worker.Options{ ProtocolVersion: loaded.Value.ControlPlane.ProtocolVersion, HeartbeatInterval: loaded.Value.ControlPlane.HeartbeatInterval.Value(), @@ -277,6 +281,7 @@ func runWithWorkerFactory( MaxRuntimeCounters: loaded.Value.ControlPlane.MaxRuntimeCounters, MaxSnapshotBytes: loaded.Value.ControlPlane.MaxMessageBytes, SnapshotReader: snapshotReader, + DrainTickets: drainTickets, RoutingSource: routingSource, Credentials: opened.credentials, }) diff --git a/internal/controller/worker/grpc_handler.go b/internal/controller/worker/grpc_handler.go index e04dc29..42435c6 100644 --- a/internal/controller/worker/grpc_handler.go +++ b/internal/controller/worker/grpc_handler.go @@ -99,6 +99,14 @@ func (handler *GRPCHandler) WatchSnapshots(request *controlplanev1.WatchSnapshot if err := handler.issueSnapshot(stream.Context(), request.GetWorkerId(), request.GetSessionId(), snapshot); err != nil { return grpcError(err) } + if binder, ok := handler.service.(interface { + BindDrainBarriers(context.Context, string, string, workerruntime.SnapshotReference, []string) error + }); ok { + reference := snapshotReference(request.GetWorkerId(), snapshot) + if err := binder.BindDrainBarriers(stream.Context(), request.GetWorkerId(), request.GetSessionId(), reference, snapshotProxyIDs(snapshot)); err != nil { + return grpcError(err) + } + } if err := stream.Send(&controlplanev1.SnapshotEnvelope{Payload: &controlplanev1.SnapshotEnvelope_Full{Full: snapshot}}); err != nil { return err } @@ -114,11 +122,25 @@ func (handler *GRPCHandler) issueSnapshot(ctx context.Context, workerID, session !snapshot.GetValidUntil().AsTime().After(time.Now().UTC()) { return ErrInvalidCommand } + return handler.service.IssueSnapshot(ctx, sessionID, snapshotReference(workerID, snapshot)) +} + +func snapshotReference(workerID string, snapshot *controlplanev1.WorkerSnapshot) workerruntime.SnapshotReference { var checksum [sha256.Size]byte copy(checksum[:], snapshot.GetChecksum()) - return handler.service.IssueSnapshot(ctx, sessionID, workerruntime.SnapshotReference{ + return workerruntime.SnapshotReference{ WorkerID: workerID, Version: snapshot.GetVersion(), OwnershipEpoch: snapshot.GetOwnershipEpoch(), Checksum: checksum, - }) + } +} + +func snapshotProxyIDs(snapshot *controlplanev1.WorkerSnapshot) []string { + proxyIDs := make([]string, 0, len(snapshot.GetProxies())) + for _, proxy := range snapshot.GetProxies() { + if proxy != nil { + proxyIDs = append(proxyIDs, proxy.GetId()) + } + } + return proxyIDs } func resetSnapshotExpiryTimer(timer *time.Timer, deadline time.Time) (*time.Timer, <-chan time.Time) { diff --git a/internal/controller/worker/grpc_handler_test.go b/internal/controller/worker/grpc_handler_test.go index 84f6025..97c9728 100644 --- a/internal/controller/worker/grpc_handler_test.go +++ b/internal/controller/worker/grpc_handler_test.go @@ -171,6 +171,28 @@ func TestGRPCHandlerStreamsIssuedFullSnapshots(t *testing.T) { } } +func TestGRPCHandlerBindsDrainBarriersAfterIssuingExcludedSnapshot(t *testing.T) { + checksum := make([]byte, 32) + checksum[0] = 1 + service := &drainBindingServiceStub{grpcServiceStub: grpcServiceStub{}} + client, cleanup := grpcWorkerClient(t, service, allowIdentity{}, snapshotSourceStub{snapshots: []*controlplanev1.WorkerSnapshot{{ + Version: 3, OwnershipEpoch: 9, Checksum: checksum, + GeneratedAt: timestamppb.New(time.Now()), ValidUntil: timestamppb.New(time.Now().Add(time.Minute)), + Proxies: []*controlplanev1.OwnedProxy{{Id: "proxy-present"}}, + }}}) + defer cleanup() + stream, err := client.WatchSnapshots(context.Background(), &controlplanev1.WatchSnapshotsRequest{WorkerId: "worker-a", SessionId: "session-a"}) + if err != nil { + t.Fatalf("WatchSnapshots(): %v", err) + } + if _, err := stream.Recv(); err != nil { + t.Fatalf("Recv(): %v", err) + } + if !service.boundAfterIssue || service.boundReference.Version != 3 || len(service.presentProxyIDs) != 1 || service.presentProxyIDs[0] != "proxy-present" { + t.Fatalf("BindDrainBarriers() = afterIssue:%t reference:%+v present:%v", service.boundAfterIssue, service.boundReference, service.presentProxyIDs) + } +} + func TestGRPCHandlerClosesSnapshotStreamAtValidityDeadline(t *testing.T) { checksum := make([]byte, 32) checksum[0] = 1 @@ -255,6 +277,20 @@ type grpcServiceStub struct { issueErr error } +type drainBindingServiceStub struct { + grpcServiceStub + boundAfterIssue bool + boundReference workerruntime.SnapshotReference + presentProxyIDs []string +} + +func (stub *drainBindingServiceStub) BindDrainBarriers(_ context.Context, _ string, _ string, reference workerruntime.SnapshotReference, proxyIDs []string) error { + stub.boundAfterIssue = stub.issued.Version == reference.Version + stub.boundReference = reference + stub.presentProxyIDs = append([]string(nil), proxyIDs...) + return nil +} + func (stub *grpcServiceStub) Register(context.Context, RegisterCommand) (Registration, error) { return stub.registration, stub.registerErr } diff --git a/internal/controller/worker/service.go b/internal/controller/worker/service.go index aed3818..3aa7120 100644 --- a/internal/controller/worker/service.go +++ b/internal/controller/worker/service.go @@ -60,6 +60,7 @@ type Options struct { MaxRuntimeCounters int MaxSnapshotBytes int SnapshotReader ownershipDomain.SnapshotReader + DrainTickets ownershipDomain.DrainTicketStore RoutingSource RoutingSource Credentials platformCredentials.Store SessionID func() (string, error) @@ -122,6 +123,7 @@ func (service *service) IssueSnapshot(ctx context.Context, sessionID string, ref type service struct { store workerruntime.ControlStore outcomes workerruntime.OutcomeWriter + drains ownershipDomain.DrainTicketStore options Options snapshots SnapshotSource } @@ -135,7 +137,7 @@ func NewService(store workerruntime.ControlStore, options Options) (Service, err if options.SessionID == nil { options.SessionID = randomSessionID } - result := &service{store: store, options: options} + result := &service{store: store, drains: options.DrainTickets, options: options} result.outcomes, _ = store.(workerruntime.OutcomeWriter) if options.SnapshotReader != nil { var routing []RoutingSource @@ -165,6 +167,54 @@ func NewService(store workerruntime.ControlStore, options Options) (Service, err return result, nil } +// BindDrainBarriers records the issued full snapshot which excludes each +// pending proxy. The handler invokes it only after the snapshot reference is +// durably issued for the current session. +func (service *service) BindDrainBarriers( + ctx context.Context, + workerID, sessionID string, + reference workerruntime.SnapshotReference, + presentProxyIDs []string, +) error { + if service == nil || service.drains == nil { + return nil + } + if ctx == nil || !workerruntime.ValidIdentifier(workerID) || !workerruntime.ValidIdentifier(sessionID) { + return ErrInvalidCommand + } + normalized, err := workerruntime.NormalizeSnapshotReference(reference) + if err != nil || normalized.WorkerID != workerID { + return errors.Join(ErrInvalidCommand, err) + } + present := make(map[string]struct{}, len(presentProxyIDs)) + for _, proxyID := range presentProxyIDs { + if !workerruntime.ValidIdentifier(proxyID) { + return ErrInvalidCommand + } + if _, exists := present[proxyID]; exists { + return ErrInvalidCommand + } + present[proxyID] = struct{}{} + } + tickets, err := service.drains.PendingDrains(ctx, workerID, service.options.MaxRuntimeCounters) + if err != nil { + return classifyStoreError(err) + } + for _, ticket := range tickets { + if _, exists := present[ticket.ProxyID]; exists { + continue + } + ticket.Barrier = ownershipDomain.SnapshotBarrier{ + SessionID: sessionID, Version: normalized.Version, OwnershipEpoch: normalized.OwnershipEpoch, + Checksum: normalized.Checksum, + } + if err := service.drains.BindDrainBarrier(ctx, ticket); err != nil { + return classifyStoreError(err) + } + } + return nil +} + func (service *service) SnapshotSource() SnapshotSource { if service == nil { return nil diff --git a/internal/domain/activitypool/contracttest/contract.go b/internal/domain/activitypool/contracttest/contract.go index e016a4d..9938dd9 100644 --- a/internal/domain/activitypool/contracttest/contract.go +++ b/internal/domain/activitypool/contracttest/contract.go @@ -299,6 +299,17 @@ func runOwnershipContract(t *testing.T, factory Factory) { pending[0].AssignmentEpoch != assigned.Epoch || pending[0].RequiredSnapshotEpoch <= assigned.Epoch { t.Fatalf("PendingDrains() = %+v, %v", pending, err) } + pending[0].Barrier = ownershipDomain.SnapshotBarrier{ + SessionID: "session-a", Version: 1, OwnershipEpoch: pending[0].RequiredSnapshotEpoch, + Checksum: [32]byte{1}, + } + if err := store.BindDrainBarrier(context.Background(), pending[0]); err != nil { + t.Fatalf("BindDrainBarrier(): %v", err) + } + bound, err := store.PendingDrains(context.Background(), "worker-a", 1) + if err != nil || len(bound) != 1 || bound[0].Barrier != pending[0].Barrier { + t.Fatalf("PendingDrains(bound) = %+v, %v", bound, 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) } diff --git a/internal/domain/activitypool/pool.go b/internal/domain/activitypool/pool.go index 6daa87c..9a27206 100644 --- a/internal/domain/activitypool/pool.go +++ b/internal/domain/activitypool/pool.go @@ -929,6 +929,35 @@ func (p *MemoryPool) PendingDrains(ctx context.Context, workerID string, limit i return tickets, nil } +func (p *MemoryPool) BindDrainBarrier(ctx context.Context, ticket ownershipDomain.DrainTicket) error { + if err := ownershipContextError(ctx); err != nil { + return err + } + if p == nil || !validDrainTicket(ticket) || !validSnapshotBarrier(ticket.Barrier) || + ticket.Barrier.OwnershipEpoch < ticket.RequiredSnapshotEpoch { + return ownershipDomain.ErrInvalidDrainTicket + } + p.mu.Lock() + defer p.mu.Unlock() + if err := ctx.Err(); err != nil { + return err + } + current, exists := p.drains[ticket.ProxyID] + assignment, owned := p.ownership[ticket.ProxyID] + if !exists || !owned || !assignment.Draining || assignment.WorkerID != ticket.WorkerID || + assignment.Epoch != ticket.AssignmentEpoch || current.ProxyID != ticket.ProxyID || + current.WorkerID != ticket.WorkerID || current.AssignmentEpoch != ticket.AssignmentEpoch || + current.RequiredSnapshotEpoch != ticket.RequiredSnapshotEpoch { + return ownershipDomain.ErrStaleAssignment + } + if barrierAfter(current.Barrier, ticket.Barrier) || barrierEqual(current.Barrier, ticket.Barrier) { + return nil + } + current.Barrier = ticket.Barrier + p.drains[ticket.ProxyID] = current + return 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 @@ -1025,6 +1054,27 @@ func ownershipContextError(ctx context.Context) error { return ctx.Err() } +func validDrainTicket(ticket ownershipDomain.DrainTicket) bool { + return ticket.ProxyID != "" && ticket.WorkerID != "" && ticket.AssignmentEpoch > 0 && + ticket.RequiredSnapshotEpoch > ticket.AssignmentEpoch +} + +func validSnapshotBarrier(barrier ownershipDomain.SnapshotBarrier) bool { + return barrier.SessionID != "" && barrier.Version > 0 && barrier.OwnershipEpoch > 0 && + barrier.Checksum != [sha256.Size]byte{} +} + +func barrierAfter(current, next ownershipDomain.SnapshotBarrier) bool { + if current.OwnershipEpoch != next.OwnershipEpoch { + return current.OwnershipEpoch > next.OwnershipEpoch + } + return current.Version > next.Version +} + +func barrierEqual(left, right ownershipDomain.SnapshotBarrier) bool { + return left == right +} + func (p *MemoryPool) purgeExpiredLocked(now time.Time) int { removed := 0 for key, entry := range p.entries { diff --git a/internal/domain/ownership/ownership.go b/internal/domain/ownership/ownership.go index 849840a..1f9f7c0 100644 --- a/internal/domain/ownership/ownership.go +++ b/internal/domain/ownership/ownership.go @@ -2,6 +2,7 @@ package ownership import ( "context" + "crypto/sha256" "errors" "time" ) @@ -33,6 +34,14 @@ type DrainTicket struct { WorkerID string AssignmentEpoch uint64 RequiredSnapshotEpoch uint64 + Barrier SnapshotBarrier +} + +type SnapshotBarrier struct { + SessionID string + Version uint64 + OwnershipEpoch uint64 + Checksum [sha256.Size]byte } // DrainTicketStore exposes bounded, Worker-local pending drain tickets. It @@ -40,6 +49,7 @@ type DrainTicket struct { // prove both an acknowledged exclusion snapshot and zero live counters. type DrainTicketStore interface { PendingDrains(context.Context, string, int) ([]DrainTicket, error) + BindDrainBarrier(context.Context, DrainTicket) error } // Repository is the shared authority for ownership changes. Implementations diff --git a/progress.md b/progress.md index a4e5a46..d64574d 100644 --- a/progress.md +++ b/progress.md @@ -26,8 +26,9 @@ 有效刷新被错误拒绝。这是后续自动 Drain 编排的必要前提。 - Drain 首次发起现在原子推进 ownership epoch，并写入由 Proxy、Worker、原 assignment epoch 与所需 Snapshot epoch 组成的待绑定 Ticket；Redis 使用 `drain-tickets` 与按 - Worker 的 `worker-draining` 索引，ACK、过期和硬删除会清理 Ticket。完整快照排除、 - ACK 与 Runtime 零计数的原子自动完成仍在后续切片，尚未标记为完成。 + Worker 的 `worker-draining` 索引，ACK、过期和硬删除会清理 Ticket。Worker Handler + 在持久登记完整 Snapshot 引用后，核验代理确实不在完整视图内，再绑定 Ticket 的 + session/version/epoch/checksum 屏障；ACK 与 Runtime 零计数的原子自动完成仍在后续切片。 ## 2026-07-30 diff --git a/task_plan.md b/task_plan.md index 0cd5181..b6365f1 100644 --- a/task_plan.md +++ b/task_plan.md @@ -41,8 +41,8 @@ 通过 Controller 后台的有界回收清理无 Worker ownership 的持续异常 Proxy；拥有 Worker 的候选延后到既有 Drain/ACK 清除所有权后再处理。 15. [进行中] 收敛 Worker 发布生命周期：已完成权威 Proxy/Routing 的持续完整快照 - 刷新，以及 Drain 发起时的持久化 Ticket、Worker 待绑定索引和 ownership epoch 推进； - 后续在完整快照排除、ACK 与 Runtime 零计数之间补齐原子自动完成编排，使配置停用、 + 刷新，以及 Drain 发起时的持久化 Ticket、Worker 待绑定索引、ownership epoch 推进和 + 完整排除 Snapshot 屏障绑定；后续在 ACK 与 Runtime 零计数之间补齐原子自动完成编排，使配置停用、 健康淘汰与 Snapshot 撤销形成可观测闭环。 ## 串并行关系