test: cover worker control plane bootstrap flow

internal/controller/bootstrap/bootstrap_integration_test.go

@@ -4,7 +4,9 @@ package bootstrap
 
 import (
 	"context"
+	"crypto/sha256"
 	"errors"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -22,7 +24,14 @@ import (
 	"proxy-pool/internal/controller/pool"
 	"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"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/protobuf/types/known/timestamppb"
+	controlplanev1 "proxy-pool/gen/controlplane/v1"
 )
 
 func TestProductionBootstrapOpensBothStoresAndCommitsStartupConfiguration(t *testing.T) {
@@ -80,6 +89,95 @@ func TestProductionBootstrapOpensBothStoresAndCommitsStartupConfiguration(t *tes
 	}
 }
 
+func TestProductionBootstrapServesWorkerControlPlane(t *testing.T) {
+	postgresURL := os.Getenv("PROXY_POOL_TEST_POSTGRES_URL")
+	redisURL := os.Getenv("PROXY_POOL_TEST_REDIS_URL")
+	if postgresURL == "" || redisURL == "" {
+		t.Skip("PROXY_POOL_TEST_POSTGRES_URL and PROXY_POOL_TEST_REDIS_URL are required")
+	}
+	providerServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		_, _ = writer.Write([]byte("http://192.0.2.10:8080"))
+	}))
+	defer providerServer.Close()
+	namespace := "controller-worker-it-" + strconv.FormatInt(time.Now().UnixNano(), 10)
+	source := strings.ReplaceAll(bootstrapTestConfig, "postgres://fixture", postgresURL)
+	source = strings.ReplaceAll(source, "redis://fixture", redisURL)
+	source = strings.ReplaceAll(source, "https://provider.invalid/proxies", providerServer.URL)
+	source += `
+controlPlane:
+  enabled: true
+  listen: 127.0.0.1:0
+  protocolVersion: 1
+  heartbeatInterval: 10s
+  sessionTTL: 30s
+  maxStaleAge: 10s
+  maxMessageBytes: 1048576
+  maxRuntimeCounters: 100
+  maxConcurrentStreams: 10
+  tls: {mode: disabled}
+`
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	workerFactory := &integrationWorkerRuntimeFactory{ready: make(chan struct{})}
+	result := make(chan error, 1)
+	go func() {
+		result <- runWithWorkerFactory(ctx, Options{
+			ConfigPath: "controller.yaml", Resolver: &memoryResolver{files: map[string][]byte{"controller.yaml": []byte(source)}},
+			Now: time.Now, FingerprintKey: bootstrapTestFingerprintKey,
+		}, &productionInfrastructure{namespace: namespace}, integrationBlockingRuntimeFactory{}, workerFactory)
+	}()
+	select {
+	case <-workerFactory.ready:
+	case <-time.After(5 * time.Second):
+		t.Fatal("worker control plane did not start")
+	}
+
+	connection, err := grpc.NewClient(workerFactory.address, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(): %v", err)
+	}
+	defer connection.Close()
+	client := controlplanev1.NewWorkerControlPlaneClient(connection)
+	requestCtx, requestCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer requestCancel()
+	registration, err := client.RegisterWorker(requestCtx, &controlplanev1.RegisterWorkerRequest{
+		WorkerId: "worker-a", InstanceId: "instance-a", Zone: "zone-a", SupportedProtocolVersion: 1,
+	})
+	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,
+	}
+	if err := store.RecordIssuedSnapshot(requestCtx, reference, time.Minute); err != nil {
+		t.Fatalf("RecordIssuedSnapshot(): %v", err)
+	}
+	if _, err := client.AcknowledgeSnapshot(requestCtx, &controlplanev1.AcknowledgeSnapshotRequest{
+		WorkerId: "worker-a", SessionId: registration.GetSessionId(), Version: 1,
+		OwnershipEpoch: registration.GetOwnershipEpoch(), Checksum: checksum[:], 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(),
+	})
+	if err != nil || runtime.GetRequireFullSnapshot() || runtime.GetAcceptedOwnershipEpoch() != registration.GetOwnershipEpoch() {
+		t.Fatalf("ReportRuntime() = %+v, %v", runtime, err)
+	}
+	cancel()
+	select {
+	case err := <-result:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("run() error = %v, want context cancellation", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("controller did not stop")
+	}
+}
+
 type integrationRuntimeFactory struct {
 	status          admin.Status
 	readyStatus     int
@@ -206,3 +304,62 @@ type integrationRunner struct {
 }
 
 func (runner integrationRunner) Run(ctx context.Context) error { return runner.run(ctx) }
+
+type integrationBlockingRuntimeFactory struct{}
+
+func (integrationBlockingRuntimeFactory) New(
+	_ *config.Config,
+	_ controllerRuntime.Dependencies,
+	_ controllerRuntime.Options,
+) (controllerRunner, error) {
+	return integrationRunner{run: func(ctx context.Context) error {
+		<-ctx.Done()
+		return ctx.Err()
+	}}, nil
+}
+
+type integrationWorkerRuntimeFactory struct {
+	address string
+	ready   chan struct{}
+}
+
+func (factory *integrationWorkerRuntimeFactory) New(
+	controlPlane config.ControlPlane,
+	service controllerWorker.Service,
+) (controllerRunner, error) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		return nil, err
+	}
+	server, err := controllerWorker.NewServer(controlPlane, service, controllerWorker.DefaultServerOptions())
+	if err != nil {
+		_ = listener.Close()
+		return nil, err
+	}
+	factory.address = listener.Addr().String()
+	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
+}