feat: run worker grpc control plane in controller

internal/controller/bootstrap/bootstrap.go

@@ -19,8 +19,10 @@ import (
 	"proxy-pool/internal/controller/pool"
 	"proxy-pool/internal/controller/provider"
 	controllerRuntime "proxy-pool/internal/controller/runtime"
+	"proxy-pool/internal/controller/worker"
 	"proxy-pool/internal/domain/activitypool"
 	extractionDomain "proxy-pool/internal/domain/extraction"
+	"proxy-pool/internal/domain/workerruntime"
 	"proxy-pool/internal/platform/admission"
 	"proxy-pool/internal/platform/credentials"
 	"proxy-pool/internal/platform/httpserver"
@@ -59,6 +61,7 @@ type ports struct {
 	coordinator      provider.Coordinator
 	credentials      credentials.Store
 	providerResults  provider.ResultRecorder
+	workerStore      workerruntime.ControlStore
 	close            func() error
 }
 
@@ -74,15 +77,29 @@ type runtimeFactory interface {
 	New(*config.Config, controllerRuntime.Dependencies, controllerRuntime.Options) (controllerRunner, error)
 }
 
+type workerRuntimeFactory interface {
+	New(config.ControlPlane, worker.Service) (controllerRunner, error)
+}
+
 func Run(ctx context.Context, options Options) error {
-	return run(ctx, options, &productionInfrastructure{
+	return runWithWorkerFactory(ctx, options, &productionInfrastructure{
 		holderID: options.HolderID, namespace: options.RedisNamespace,
-	}, productionRuntimeFactory{})
+	}, productionRuntimeFactory{}, productionWorkerRuntimeFactory{})
 }
 
 func run(ctx context.Context, options Options, infrastructure infrastructure, factory runtimeFactory) (resultErr error) {
+	return runWithWorkerFactory(ctx, options, infrastructure, factory, productionWorkerRuntimeFactory{})
+}
+
+func runWithWorkerFactory(
+	ctx context.Context,
+	options Options,
+	infrastructure infrastructure,
+	factory runtimeFactory,
+	workerFactory workerRuntimeFactory,
+) (resultErr error) {
 	if ctx == nil || strings.TrimSpace(options.ConfigPath) != options.ConfigPath || options.ConfigPath == "" ||
-		nilInterface(options.Resolver) || nilInterface(infrastructure) || nilInterface(factory) {
+		nilInterface(options.Resolver) || nilInterface(infrastructure) || nilInterface(factory) || nilInterface(workerFactory) {
 		return ErrInvalidOptions
 	}
 	if err := ctx.Err(); err != nil {
@@ -200,7 +217,7 @@ func run(ctx context.Context, options Options, infrastructure infrastructure, fa
 		dependencies.MetricsHandler = handler
 	}
 
-	runners := make([]lifecycle.Runner, 0, 2)
+	runners := make([]lifecycle.Runner, 0, 3)
 	if hasHTTPRuntime(loaded.Value) {
 		runner, err := factory.New(configurationStore.Current(), dependencies, controllerRuntime.Options{HTTP: options.HTTP})
 		if err != nil {
@@ -211,6 +228,29 @@ func run(ctx context.Context, options Options, infrastructure infrastructure, fa
 		}
 		runners = append(runners, runner)
 	}
+	if loaded.Value.ControlPlane.Enabled {
+		if nilInterface(opened.workerStore) {
+			return errors.Join(ErrStartup, ErrInvalidOptions)
+		}
+		service, serviceErr := worker.NewService(opened.workerStore, worker.Options{
+			ProtocolVersion:    loaded.Value.ControlPlane.ProtocolVersion,
+			HeartbeatInterval:  loaded.Value.ControlPlane.HeartbeatInterval.Value(),
+			SessionTTL:         loaded.Value.ControlPlane.SessionTTL.Value(),
+			MaxStaleAge:        loaded.Value.ControlPlane.MaxStaleAge.Value(),
+			MaxRuntimeCounters: loaded.Value.ControlPlane.MaxRuntimeCounters,
+		})
+		if serviceErr != nil {
+			return fmt.Errorf("%w: build Worker control service: %w", ErrStartup, serviceErr)
+		}
+		runner, runnerErr := workerFactory.New(loaded.Value.ControlPlane, service)
+		if runnerErr != nil {
+			return fmt.Errorf("%w: build Worker control server: %w", ErrStartup, runnerErr)
+		}
+		if nilInterface(runner) {
+			return errors.Join(ErrStartup, ErrInvalidOptions)
+		}
+		runners = append(runners, runner)
+	}
 	runners = append(runners, supervisor)
 	group, err := lifecycle.NewGroup(runners...)
 	if err != nil {
@@ -298,3 +338,9 @@ func (productionRuntimeFactory) New(
 ) (controllerRunner, error) {
 	return controllerRuntime.New(configuration, dependencies, options)
 }
+
+type productionWorkerRuntimeFactory struct{}
+
+func (productionWorkerRuntimeFactory) New(controlPlane config.ControlPlane, service worker.Service) (controllerRunner, error) {
+	return worker.NewServer(controlPlane, service, worker.DefaultServerOptions())
+}

internal/controller/bootstrap/bootstrap_test.go

@@ -12,10 +12,12 @@ 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/activitypool"
 	"proxy-pool/internal/domain/adminstate"
 	extractionDomain "proxy-pool/internal/domain/extraction"
 	"proxy-pool/internal/domain/upstream"
+	"proxy-pool/internal/domain/workerruntime"
 	"proxy-pool/internal/platform/admission"
 	"proxy-pool/internal/platform/credentials"
 )
@@ -170,6 +172,52 @@ func TestRunSupportsProviderOnlyConfigurationWithoutHTTPRuntime(t *testing.T) {
 	}
 }
 
+func TestRunRequiresWorkerControlStoreWhenControlPlaneEnabled(t *testing.T) {
+	source := bootstrapControlPlaneConfig()
+	credentialStore, err := credentials.NewMemoryStore(10)
+	if err != nil {
+		t.Fatalf("NewMemoryStore(): %v", err)
+	}
+	infrastructure := &stubInfrastructure{ports: ports{
+		activity: &stubActivityStore{}, coordinator: coordinatorStub{}, credentials: credentialStore, close: func() error { return nil },
+	}}
+	err = run(context.Background(), Options{
+		ConfigPath: "controller.yaml", Resolver: &memoryResolver{files: map[string][]byte{"controller.yaml": []byte(source)}}, Now: time.Now,
+	}, infrastructure, &recordingRuntimeFactory{})
+	if !errors.Is(err, ErrStartup) || !errors.Is(err, ErrInvalidOptions) {
+		t.Fatalf("run() error = %v, want startup invalid options", err)
+	}
+}
+
+func TestRunStartsWorkerControlPlaneWithoutHTTPRuntime(t *testing.T) {
+	store, err := workerruntime.NewMemoryStore(time.Now)
+	if err != nil {
+		t.Fatalf("NewMemoryStore(): %v", err)
+	}
+	credentialStore, err := credentials.NewMemoryStore(10)
+	if err != nil {
+		t.Fatalf("NewMemoryStore(): %v", err)
+	}
+	infrastructure := &stubInfrastructure{ports: ports{
+		activity: &stubActivityStore{}, workerStore: store, coordinator: coordinatorStub{}, credentials: credentialStore, close: func() error { return nil },
+	}}
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	workerFactory := &recordingWorkerRuntimeFactory{runner: runnerFunc(func(ctx context.Context) error {
+		<-ctx.Done()
+		return ctx.Err()
+	})}
+	err = runWithWorkerFactory(ctx, Options{
+		ConfigPath: "controller.yaml", Resolver: &memoryResolver{files: map[string][]byte{"controller.yaml": []byte(bootstrapControlPlaneConfig())}}, Now: time.Now,
+	}, infrastructure, &recordingRuntimeFactory{}, workerFactory)
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("run() error = %v, want context deadline exceeded", err)
+	}
+	if workerFactory.controlPlane.Listen != "127.0.0.1:0" || workerFactory.service == nil {
+		t.Fatalf("worker runtime factory = controlPlane:%+v service:%T", workerFactory.controlPlane, workerFactory.service)
+	}
+}
+
 func TestRunAdminDisableStopsActiveProviderRuntime(t *testing.T) {
 	state := adminstate.NewMemoryStore()
 	credentialStore, err := credentials.NewMemoryStore(200)
@@ -332,6 +380,22 @@ func (factory runtimeFactoryFunc) New(
 	return factory(configuration, dependencies, options)
 }
 
+type recordingWorkerRuntimeFactory struct {
+	controlPlane config.ControlPlane
+	service      controllerWorker.Service
+	runner       controllerRunner
+	err          error
+}
+
+func (factory *recordingWorkerRuntimeFactory) New(
+	controlPlane config.ControlPlane,
+	service controllerWorker.Service,
+) (controllerRunner, error) {
+	factory.controlPlane = controlPlane
+	factory.service = service
+	return factory.runner, factory.err
+}
+
 type readyStub struct{}
 
 func (readyStub) Ready(context.Context) error { return nil }
@@ -453,3 +517,22 @@ upstreams:
       urls: [https://example.invalid/health]
   provider-b: *upstream
 `
+
+func bootstrapControlPlaneConfig() string {
+	source := strings.ReplaceAll(bootstrapTestConfig, "distribution:\n  enabled: true", "distribution:\n  enabled: false")
+	source = strings.ReplaceAll(source, "admin:\n  enabled: true", "admin:\n  enabled: false")
+	source = strings.ReplaceAll(source, "metrics:\n  enabled: true", "metrics:\n  enabled: false")
+	return 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}
+`
+}

internal/controller/bootstrap/infrastructure.go

@@ -102,7 +102,7 @@ func (infrastructure *productionInfrastructure) Open(
 	}
 
 	providersEnabled := hasEnabledUpstream(configuration)
-	if configuration.Distribution.Enabled || configuration.Admin.Enabled || providersEnabled {
+	if configuration.Distribution.Enabled || configuration.Admin.Enabled || providersEnabled || configuration.ControlPlane.Enabled {
 		if strings.TrimSpace(configuration.Storage.RedisURL) == "" {
 			return ports{}, ErrRedisConfiguration
 		}
@@ -123,7 +123,7 @@ func (infrastructure *productionInfrastructure) Open(
 			Credentials:        credentialStore,
 			OperationTTL:       redisOperationTTL,
 			MaxCandidateScan:   candidateScan(configuration),
-			MaxRuntimeCounters: credentialCapacity(configuration),
+			MaxRuntimeCounters: runtimeCounterCapacity(configuration),
 			MaxInventoryScan:   maxInventoryScan(configuration),
 			CleanupLimit:       redisCleanupLimit,
 		})
@@ -131,6 +131,7 @@ func (infrastructure *productionInfrastructure) Open(
 			return ports{}, err
 		}
 		opened.activity = adapter
+		opened.workerStore = adapter
 		opened.readiness = redisReadiness{client: redisClient}
 		opened.credentials = credentialStore
 		if configuration.Distribution.Enabled {
@@ -203,7 +204,7 @@ func selectMetricsReadiness(
 	configuration *config.Config,
 	admin, activity platformMetrics.ReadinessChecker,
 ) platformMetrics.ReadinessChecker {
-	if configuration.Distribution.Enabled || hasEnabledUpstream(configuration) {
+	if configuration.Distribution.Enabled || configuration.ControlPlane.Enabled || hasEnabledUpstream(configuration) {
 		return activity
 	}
 	if configuration.Admin.Enabled {
@@ -275,6 +276,14 @@ func credentialCapacity(configuration *config.Config) int {
 	return capacity
 }
 
+func runtimeCounterCapacity(configuration *config.Config) int {
+	capacity := credentialCapacity(configuration)
+	if configuration != nil && configuration.ControlPlane.Enabled && configuration.ControlPlane.MaxRuntimeCounters > capacity {
+		return configuration.ControlPlane.MaxRuntimeCounters
+	}
+	return capacity
+}
+
 func providerCredentialCapacity(configuration *config.Config) int {
 	capacity := 0
 	maximum := int(^uint(0) >> 1)

internal/controller/bootstrap/infrastructure_test.go

@@ -50,6 +50,13 @@ func TestProductionInfrastructureRejectsInvalidStorageWithoutLeakingURLs(t *test
 	if !errors.Is(err, ErrRedisConfiguration) || strings.Contains(err.Error(), redisSecret) {
 		t.Fatalf("Open(invalid Redis) error = %v", err)
 	}
+	_, err = (&productionInfrastructure{}).Open(context.Background(), &config.Config{
+		ControlPlane: config.ControlPlane{Enabled: true},
+		Storage:      config.Storage{RedisURL: "redis://user:" + redisSecret + "@%zz"},
+	})
+	if !errors.Is(err, ErrRedisConfiguration) || strings.Contains(err.Error(), redisSecret) {
+		t.Fatalf("Open(control plane invalid Redis) error = %v", err)
+	}
 }
 
 func TestNewDistributionAdmitterPassesConfiguredLimits(t *testing.T) {
@@ -124,6 +131,19 @@ func TestSelectMetricsReadinessUsesAdminStoresWithoutDistribution(t *testing.T)
 	}
 }
 
+func TestSelectMetricsReadinessUsesRedisForControlPlane(t *testing.T) {
+	t.Parallel()
+	wantErr := errors.New("redis unavailable")
+	selected := selectMetricsReadiness(
+		&config.Config{ControlPlane: config.ControlPlane{Enabled: true}},
+		readinessFunc(func(context.Context) error { return nil }),
+		readinessFunc(func(context.Context) error { return wantErr }),
+	)
+	if err := selected.Ready(context.Background()); !errors.Is(err, wantErr) {
+		t.Fatalf("Ready() error = %v, want %v", err, wantErr)
+	}
+}
+
 type readinessFunc func(context.Context) error
 
 func (function readinessFunc) Ready(ctx context.Context) error { return function(ctx) }
@@ -143,6 +163,14 @@ func TestProductionRedisSizingUsesConfigurationBounds(t *testing.T) {
 	if got := credentialCapacity(configuration); got != 5_000 {
 		t.Fatalf("credentialCapacity() = %d, want 5000", got)
 	}
+	configuration.ControlPlane = config.ControlPlane{Enabled: true, MaxRuntimeCounters: 100_000}
+	if got := runtimeCounterCapacity(configuration); got != 100_000 {
+		t.Fatalf("runtimeCounterCapacity(control plane) = %d, want 100000", got)
+	}
+	configuration.ControlPlane.Enabled = false
+	if got := runtimeCounterCapacity(configuration); got != 5_000 {
+		t.Fatalf("runtimeCounterCapacity(disabled control plane) = %d, want 5000", got)
+	}
 	if got := providerCredentialCapacity(configuration); got != 158_000 {
 		t.Fatalf("providerCredentialCapacity() = %d, want 158000", got)
 	}