diff --git a/README.md b/README.md index d08d02e..57932ec 100644 --- a/README.md +++ b/README.md @@ -81,8 +81,8 @@ flowchart LR Redis 会话栅栏。 - **部分完成**：Gateway 传输与调度组件、Snapshot 本地存储、Worker ownership 与运行态领域组件、Docker Compose/Kubernetes 静态部署清单和 protobuf 契约。 -- **待完成**：Worker Snapshot 下发流、Gateway 快照客户端、Outcome 上报、Checker - 调度与健康状态链、完整 Routing 运行链，以及 loadgen 和代表性集群压测。 +- **待完成**：Worker 权威 Proxy/Routing Snapshot 发布、Gateway 进程装配、Outcome + 上报、Checker 调度与健康状态链、完整 Routing 运行链，以及 loadgen 和代表性集群压测。 检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表 端到端拓扑已经完成；`100,000 QPS` 仍只是待验证的集群设计目标。 diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index 3180075..5eb553f 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -25,9 +25,11 @@ Controller 已实现并验证 `RegisterWorker`、`AcknowledgeSnapshot` 和 监听允许明文 fixture 模式。单消息大小、并发流数和 gRPC keepalive 策略由 `controlPlane` 配置限定。 -`WatchSnapshots` 和 `ReportOutcomes` 当前明确返回 `Unimplemented`。 -Snapshot payload/stream、Gateway 客户端、Outcome 与 Checker 闭环尚未实现； -`100,000 QPS` 仍是未验证的设计目标。 +`WatchSnapshots` 已在 Register 后发送与当前 ownership epoch 对应的基础完整快照， +Gateway 校验后 ACK 并开始 Runtime 心跳。当前基础快照不包含权威 Proxy 或 Routing +内容，流会保持等待后续发布；权威快照发布器、增量、Gateway 进程装配、Outcome 与 +Checker 闭环尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`；`100,000 QPS` +仍是未验证的设计目标。 ## 2. Worker 会话 diff --git a/internal/controller/bootstrap/bootstrap_integration_test.go b/internal/controller/bootstrap/bootstrap_integration_test.go index b89c7a6..8eb51ff 100644 --- a/internal/controller/bootstrap/bootstrap_integration_test.go +++ b/internal/controller/bootstrap/bootstrap_integration_test.go @@ -4,7 +4,6 @@ package bootstrap import ( "context" - "crypto/sha256" "errors" "net" "net/http" @@ -25,7 +24,6 @@ import ( "proxy-pool/internal/controller/provider" controllerRuntime "proxy-pool/internal/controller/runtime" controllerWorker "proxy-pool/internal/controller/worker" - "proxy-pool/internal/domain/workerruntime" "proxy-pool/internal/platform/credentials" "google.golang.org/grpc" @@ -146,23 +144,25 @@ controlPlane: if err != nil { t.Fatalf("RegisterWorker(): %v", err) } - store := newIntegrationWorkerStore(t, redisURL, namespace) - checksum := sha256.Sum256([]byte("snapshot-1")) - reference := workerruntime.SnapshotReference{ - WorkerID: "worker-a", Version: 1, OwnershipEpoch: registration.GetOwnershipEpoch(), Checksum: checksum, + stream, err := client.WatchSnapshots(requestCtx, &controlplanev1.WatchSnapshotsRequest{ + WorkerId: "worker-a", SessionId: registration.GetSessionId(), + }) + if err != nil { + t.Fatalf("WatchSnapshots(): %v", err) } - if err := store.RecordIssuedSnapshot(requestCtx, reference, time.Minute); err != nil { - t.Fatalf("RecordIssuedSnapshot(): %v", err) + issued, err := stream.Recv() + if err != nil || issued.GetFull() == nil || issued.GetFull().GetOwnershipEpoch() != registration.GetOwnershipEpoch() { + t.Fatalf("WatchSnapshots.Recv() = %+v, %v", issued, err) } if _, err := client.AcknowledgeSnapshot(requestCtx, &controlplanev1.AcknowledgeSnapshotRequest{ - WorkerId: "worker-a", SessionId: registration.GetSessionId(), Version: 1, - OwnershipEpoch: registration.GetOwnershipEpoch(), Checksum: checksum[:], Applied: true, + WorkerId: "worker-a", SessionId: registration.GetSessionId(), Version: issued.GetFull().GetVersion(), + OwnershipEpoch: issued.GetFull().GetOwnershipEpoch(), Checksum: issued.GetFull().GetChecksum(), Applied: true, }); err != nil { t.Fatalf("AcknowledgeSnapshot(): %v", err) } runtime, err := client.ReportRuntime(requestCtx, &controlplanev1.ReportRuntimeRequest{ - WorkerId: "worker-a", SessionId: registration.GetSessionId(), SnapshotVersion: 1, - OwnershipEpoch: registration.GetOwnershipEpoch(), ReportSequence: 1, ObservedAt: timestamppb.Now(), + WorkerId: "worker-a", SessionId: registration.GetSessionId(), SnapshotVersion: issued.GetFull().GetVersion(), + OwnershipEpoch: issued.GetFull().GetOwnershipEpoch(), ReportSequence: 1, ObservedAt: timestamppb.Now(), }) if err != nil || runtime.GetRequireFullSnapshot() || runtime.GetAcceptedOwnershipEpoch() != registration.GetOwnershipEpoch() { t.Fatalf("ReportRuntime() = %+v, %v", runtime, err) @@ -340,26 +340,3 @@ func (factory *integrationWorkerRuntimeFactory) New( close(factory.ready) return integrationRunner{run: func(ctx context.Context) error { return server.Serve(ctx, listener) }}, nil } - -func newIntegrationWorkerStore(t *testing.T, redisURL, namespace string) *redisactivity.Adapter { - t.Helper() - options, err := redis.ParseURL(redisURL) - if err != nil { - t.Fatalf("redis.ParseURL(): %v", err) - } - client := redis.NewClient(options) - t.Cleanup(func() { _ = client.Close() }) - credentialStore, err := credentials.NewMemoryStore(100) - if err != nil { - t.Fatalf("credentials.NewMemoryStore(): %v", err) - } - store, err := redisactivity.New(client, redisactivity.Options{ - Namespace: namespace, Credentials: credentialStore, OperationTTL: redisOperationTTL, - MaxCandidateScan: redisMinimumScan, MaxRuntimeCounters: 100, MaxInventoryScan: 100, - CleanupLimit: redisCleanupLimit, - }) - if err != nil { - t.Fatalf("redisactivity.New(): %v", err) - } - return store -} diff --git a/internal/controller/worker/grpc_handler_test.go b/internal/controller/worker/grpc_handler_test.go index b9070c5..37e87d8 100644 --- a/internal/controller/worker/grpc_handler_test.go +++ b/internal/controller/worker/grpc_handler_test.go @@ -100,6 +100,7 @@ type grpcServiceStub struct { func (stub *grpcServiceStub) Register(context.Context, RegisterCommand) (Registration, error) { return stub.registration, stub.registerErr } +func (stub *grpcServiceStub) CurrentOwnershipEpoch(context.Context) (uint64, error) { return 9, nil } func (stub *grpcServiceStub) IssueSnapshot(_ context.Context, reference workerruntime.SnapshotReference) error { stub.issued = reference return nil diff --git a/internal/controller/worker/server.go b/internal/controller/worker/server.go index 68054ec..abb2e12 100644 --- a/internal/controller/worker/server.go +++ b/internal/controller/worker/server.go @@ -51,6 +51,13 @@ func NewServer(controlPlane config.ControlPlane, service Service, options Server if err != nil { return nil, err } + snapshots := options.Snapshots + if snapshots == nil { + snapshots, err = NewInitialSnapshotSource(service, controlPlane.MaxStaleAge.Value(), time.Now) + if err != nil { + return nil, fmt.Errorf("%w: build initial snapshot source: %v", ErrInvalidServer, err) + } + } serverOptions = append(serverOptions, grpc.MaxRecvMsgSize(controlPlane.MaxMessageBytes), grpc.MaxSendMsgSize(controlPlane.MaxMessageBytes), @@ -61,7 +68,7 @@ func NewServer(controlPlane config.ControlPlane, service Service, options Server }), ) grpcServer := grpc.NewServer(serverOptions...) - controlplanev1.RegisterWorkerControlPlaneServer(grpcServer, NewGRPCHandler(service, identity, options.Snapshots)) + controlplanev1.RegisterWorkerControlPlaneServer(grpcServer, NewGRPCHandler(service, identity, snapshots)) return &Server{listen: controlPlane.Listen, grpcServer: grpcServer, shutdownTimeout: options.ShutdownTimeout}, nil } diff --git a/internal/controller/worker/service.go b/internal/controller/worker/service.go index 6de9ff4..0ecd9dc 100644 --- a/internal/controller/worker/service.go +++ b/internal/controller/worker/service.go @@ -60,11 +60,26 @@ type Options struct { type Service interface { Register(context.Context, RegisterCommand) (Registration, error) + CurrentOwnershipEpoch(context.Context) (uint64, error) IssueSnapshot(context.Context, workerruntime.SnapshotReference) error Acknowledge(context.Context, SnapshotAcknowledgement) error ReportRuntime(context.Context, workerruntime.Report) (RuntimeDecision, error) } +func (service *service) CurrentOwnershipEpoch(ctx context.Context) (uint64, error) { + if ctx == nil { + return 0, ErrInvalidCommand + } + if err := ctx.Err(); err != nil { + return 0, err + } + epoch, err := service.store.CurrentOwnershipEpoch(ctx) + if err != nil { + return 0, classifyStoreError(err) + } + return epoch, nil +} + func (service *service) IssueSnapshot(ctx context.Context, reference workerruntime.SnapshotReference) error { if ctx == nil { return ErrInvalidCommand diff --git a/internal/controller/worker/snapshot_source.go b/internal/controller/worker/snapshot_source.go index 23cb838..20712b9 100644 --- a/internal/controller/worker/snapshot_source.go +++ b/internal/controller/worker/snapshot_source.go @@ -3,8 +3,13 @@ package worker import ( "context" "errors" + "time" controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/controlplane/snapshotwire" + "proxy-pool/internal/domain/workerruntime" + + "google.golang.org/protobuf/types/known/timestamppb" ) var ErrSnapshotsUnavailable = errors.New("worker snapshots are unavailable") @@ -19,3 +24,52 @@ type SnapshotWatchRequest struct { type SnapshotSource interface { Watch(context.Context, SnapshotWatchRequest) (<-chan *controlplanev1.WorkerSnapshot, error) } + +type OwnershipEpochReader interface { + CurrentOwnershipEpoch(context.Context) (uint64, error) +} + +type InitialSnapshotSource struct { + epochs OwnershipEpochReader + validFor time.Duration + now func() time.Time +} + +func NewInitialSnapshotSource(epochs OwnershipEpochReader, validFor time.Duration, now func() time.Time) (*InitialSnapshotSource, error) { + if epochs == nil || validFor <= 0 || now == nil { + return nil, ErrSnapshotsUnavailable + } + return &InitialSnapshotSource{epochs: epochs, validFor: validFor, now: now}, nil +} + +func (source *InitialSnapshotSource) Watch(ctx context.Context, request SnapshotWatchRequest) (<-chan *controlplanev1.WorkerSnapshot, error) { + if source == nil || ctx == nil || !workerruntime.ValidIdentifier(request.WorkerID) || + !workerruntime.ValidIdentifier(request.SessionID) || request.LastAppliedVersion == ^uint64(0) { + return nil, ErrSnapshotsUnavailable + } + epoch, err := source.epochs.CurrentOwnershipEpoch(ctx) + if err != nil { + return nil, err + } + if epoch == 0 { + return nil, ErrSnapshotsUnavailable + } + now := source.now().UTC() + if now.IsZero() { + return nil, ErrSnapshotsUnavailable + } + full := &controlplanev1.WorkerSnapshot{ + Version: request.LastAppliedVersion + 1, OwnershipEpoch: epoch, + GeneratedAt: timestamppb.New(now), ValidUntil: timestamppb.New(now.Add(source.validFor)), + } + checksum, err := snapshotwire.Checksum(full) + if err != nil { + return nil, err + } + full.Checksum = append([]byte(nil), checksum[:]...) + updates := make(chan *controlplanev1.WorkerSnapshot, 1) + updates <- full + return updates, nil +} + +var _ SnapshotSource = (*InitialSnapshotSource)(nil) diff --git a/internal/controller/worker/snapshot_source_test.go b/internal/controller/worker/snapshot_source_test.go new file mode 100644 index 0000000..928a6ef --- /dev/null +++ b/internal/controller/worker/snapshot_source_test.go @@ -0,0 +1,40 @@ +package worker + +import ( + "context" + "testing" + "time" + + "proxy-pool/internal/controlplane/snapshotwire" +) + +func TestInitialSnapshotSourceIssuesNextFullSnapshot(t *testing.T) { + now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) + source, err := NewInitialSnapshotSource(epochReaderStub{epoch: 9}, time.Minute, func() time.Time { return now }) + if err != nil { + t.Fatalf("NewInitialSnapshotSource(): %v", err) + } + updates, err := source.Watch(context.Background(), SnapshotWatchRequest{ + WorkerID: "worker-a", SessionID: "session-a", LastAppliedVersion: 4, + }) + if err != nil { + t.Fatalf("Watch(): %v", err) + } + full := <-updates + if full.GetVersion() != 5 || full.GetOwnershipEpoch() != 9 || !full.GetValidUntil().AsTime().Equal(now.Add(time.Minute)) { + t.Fatalf("snapshot = %+v", full) + } + checksum, err := snapshotwire.Checksum(full) + if err != nil || string(checksum[:]) != string(full.GetChecksum()) { + t.Fatalf("snapshot checksum = %x, %v; want %x", full.GetChecksum(), err, checksum) + } +} + +type epochReaderStub struct { + epoch uint64 + err error +} + +func (reader epochReaderStub) CurrentOwnershipEpoch(context.Context) (uint64, error) { + return reader.epoch, reader.err +} diff --git a/internal/controlplane/snapshotwire/checksum.go b/internal/controlplane/snapshotwire/checksum.go new file mode 100644 index 0000000..4ef99ba --- /dev/null +++ b/internal/controlplane/snapshotwire/checksum.go @@ -0,0 +1,28 @@ +package snapshotwire + +import ( + "crypto/sha256" + "errors" + "fmt" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + + "google.golang.org/protobuf/proto" +) + +var ErrNilSnapshot = errors.New("nil worker snapshot") + +// Checksum is the canonical wire checksum for a full snapshot. The checksum +// field itself is excluded so both Controller and Gateway can verify it. +func Checksum(full *controlplanev1.WorkerSnapshot) ([sha256.Size]byte, error) { + if full == nil { + return [sha256.Size]byte{}, ErrNilSnapshot + } + copy := proto.Clone(full).(*controlplanev1.WorkerSnapshot) + copy.Checksum = nil + encoded, err := proto.MarshalOptions{Deterministic: true}.Marshal(copy) + if err != nil { + return [sha256.Size]byte{}, fmt.Errorf("marshal worker snapshot: %w", err) + } + return sha256.Sum256(encoded), nil +} diff --git a/internal/gateway/controlplane/watcher.go b/internal/gateway/controlplane/watcher.go index 79c9888..7199020 100644 --- a/internal/gateway/controlplane/watcher.go +++ b/internal/gateway/controlplane/watcher.go @@ -4,18 +4,17 @@ import ( "context" "crypto/sha256" "errors" - "fmt" "io" "math" "time" controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/controlplane/snapshotwire" proxyDomain "proxy-pool/internal/domain/proxy" "proxy-pool/internal/domain/workerruntime" "proxy-pool/internal/gateway/snapshot" "google.golang.org/grpc" - "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/emptypb" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -109,7 +108,7 @@ func (watcher *SnapshotWatcher) applyFull(full *controlplanev1.WorkerSnapshot) e if full == nil || full.GetVersion() == 0 || full.GetOwnershipEpoch() == 0 { return ErrInvalidSnapshotWatcher } - checksum, err := workerSnapshotChecksum(full) + checksum, err := snapshotwire.Checksum(full) if err != nil { return err } @@ -128,19 +127,6 @@ func (watcher *SnapshotWatcher) applyFull(full *controlplanev1.WorkerSnapshot) e return watcher.store.Apply(envelope) } -func workerSnapshotChecksum(full *controlplanev1.WorkerSnapshot) ([sha256.Size]byte, error) { - if full == nil { - return [sha256.Size]byte{}, ErrInvalidSnapshotWatcher - } - copy := proto.Clone(full).(*controlplanev1.WorkerSnapshot) - copy.Checksum = nil - encoded, err := proto.MarshalOptions{Deterministic: true}.Marshal(copy) - if err != nil { - return [sha256.Size]byte{}, fmt.Errorf("marshal worker snapshot: %w", err) - } - return sha256.Sum256(encoded), nil -} - func wireProxies(source []*controlplanev1.OwnedProxy) ([]proxyDomain.Proxy, error) { proxies := make([]proxyDomain.Proxy, len(source)) for index, item := range source { diff --git a/internal/gateway/controlplane/watcher_test.go b/internal/gateway/controlplane/watcher_test.go index 96fe697..7f34b34 100644 --- a/internal/gateway/controlplane/watcher_test.go +++ b/internal/gateway/controlplane/watcher_test.go @@ -8,6 +8,7 @@ import ( "time" controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/controlplane/snapshotwire" "proxy-pool/internal/gateway/snapshot" "google.golang.org/protobuf/types/known/emptypb" @@ -89,7 +90,7 @@ func (stream *snapshotStreamStub) Recv() (*controlplanev1.SnapshotEnvelope, erro func setSnapshotChecksum(t *testing.T, full *controlplanev1.WorkerSnapshot) { t.Helper() - checksum, err := workerSnapshotChecksum(full) + checksum, err := snapshotwire.Checksum(full) if err != nil { t.Fatalf("workerSnapshotChecksum(): %v", err) }