feat: publish initial worker snapshots

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` 仍只是待验证的集群设计目标。

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 会话
 

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
-}

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

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
 }
 

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

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)

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
+}

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
+}

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 {

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)
 	}