feat: add controller startup bootstrap

.github/workflows/ci.yml

@@ -48,3 +48,6 @@ jobs:
       - name: PostgreSQL admin-state contract
         shell: pwsh
         run: ./scripts/test-postgres.ps1
+      - name: Controller dual-store bootstrap
+        shell: pwsh
+        run: ./scripts/test-controller.ps1

.gitignore

@@ -9,6 +9,7 @@ coverage/
 *.out
 *.test
 *.prof
+*.exe
 .tmp-proto/
 
 # Local configuration and secrets

cmd/proxy-controller/main.go

@@ -0,0 +1,63 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+
+	"proxy-pool/internal/config"
+	"proxy-pool/internal/controller/bootstrap"
+)
+
+const configEnvironment = "PROXY_POOL_CONFIG"
+
+type environmentLookup func(string) string
+type controllerRun func(context.Context, bootstrap.Options) error
+
+func main() {
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	os.Exit(execute(ctx, os.Args[1:], os.Getenv, bootstrap.Run, os.Stderr))
+}
+
+func execute(
+	ctx context.Context,
+	args []string,
+	getenv environmentLookup,
+	run controllerRun,
+	stderr io.Writer,
+) int {
+	flags := flag.NewFlagSet("proxy-controller", flag.ContinueOnError)
+	flags.SetOutput(stderr)
+	configPath := flags.String("config", "", "configuration file path")
+	if err := flags.Parse(args); err != nil {
+		if errors.Is(err, flag.ErrHelp) {
+			return 0
+		}
+		return 2
+	}
+	if flags.NArg() != 0 {
+		_, _ = fmt.Fprintln(stderr, "proxy-controller: unexpected positional arguments")
+		return 2
+	}
+	if *configPath == "" && getenv != nil {
+		*configPath = getenv(configEnvironment)
+	}
+	if strings.TrimSpace(*configPath) != *configPath || *configPath == "" || ctx == nil || run == nil {
+		_, _ = fmt.Fprintf(stderr, "proxy-controller: -config or %s is required\n", configEnvironment)
+		return 2
+	}
+
+	err := run(ctx, bootstrap.Options{ConfigPath: *configPath, Resolver: config.OSResolver{}})
+	if err == nil || (errors.Is(err, context.Canceled) && ctx.Err() != nil) {
+		return 0
+	}
+	_, _ = fmt.Fprintf(stderr, "proxy-controller: %v\n", err)
+	return 1
+}

cmd/proxy-controller/main_test.go

@@ -0,0 +1,86 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"testing"
+
+	"proxy-pool/internal/controller/bootstrap"
+)
+
+func TestExecuteUsesFlagBeforeEnvironment(t *testing.T) {
+	t.Parallel()
+	var received bootstrap.Options
+	code := execute(context.Background(), []string{"-config", "flag.yaml"}, func(name string) string {
+		if name == configEnvironment {
+			return "environment.yaml"
+		}
+		return ""
+	}, func(_ context.Context, options bootstrap.Options) error {
+		received = options
+		return nil
+	}, &bytes.Buffer{})
+	if code != 0 || received.ConfigPath != "flag.yaml" || received.Resolver == nil {
+		t.Fatalf("execute() = %d, options = %+v", code, received)
+	}
+}
+
+func TestExecuteFallsBackToEnvironment(t *testing.T) {
+	t.Parallel()
+	var received bootstrap.Options
+	code := execute(context.Background(), nil, func(string) string { return "environment.yaml" }, func(
+		_ context.Context,
+		options bootstrap.Options,
+	) error {
+		received = options
+		return nil
+	}, &bytes.Buffer{})
+	if code != 0 || received.ConfigPath != "environment.yaml" {
+		t.Fatalf("execute() = %d, config = %q", code, received.ConfigPath)
+	}
+}
+
+func TestExecuteReturnsUsageCodeWithoutConfiguration(t *testing.T) {
+	t.Parallel()
+	called := false
+	var stderr bytes.Buffer
+	code := execute(context.Background(), nil, func(string) string { return "" }, func(
+		context.Context,
+		bootstrap.Options,
+	) error {
+		called = true
+		return nil
+	}, &stderr)
+	if code != 2 || called || stderr.Len() == 0 {
+		t.Fatalf("execute() = %d, called = %t, stderr = %q", code, called, stderr.String())
+	}
+}
+
+func TestExecuteMapsStartupFailureAndSignalCancellation(t *testing.T) {
+	t.Parallel()
+	var stderr bytes.Buffer
+	want := errors.New("startup failed")
+	code := execute(context.Background(), []string{"-config", "config.yaml"}, func(string) string { return "" }, func(
+		context.Context,
+		bootstrap.Options,
+	) error {
+		return want
+	}, &stderr)
+	if code != 1 || !bytes.Contains(stderr.Bytes(), []byte(want.Error())) {
+		t.Fatalf("execute(startup failure) = %d, stderr = %q", code, stderr.String())
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	stderr.Reset()
+	code = execute(ctx, []string{"-config", "config.yaml"}, func(string) string { return "" }, func(
+		context.Context,
+		bootstrap.Options,
+	) error {
+		return context.Canceled
+	}, &stderr)
+	if code != 0 || stderr.Len() != 0 {
+		t.Fatalf("execute(canceled) = %d, stderr = %q", code, stderr.String())
+	}
+}

deploy/compose_test.go

@@ -76,6 +76,26 @@ func TestPostgresFixtureScriptUsesDedicatedComposeProject(t *testing.T) {
 	}
 }
 
+func TestControllerFixtureScriptStartsBothStoresInDedicatedProject(t *testing.T) {
+	payload, err := os.ReadFile("../scripts/test-controller.ps1")
+	if err != nil {
+		t.Fatalf("read test-controller.ps1: %v", err)
+	}
+	script := string(payload)
+	for _, required := range []string{
+		`-p $composeProject`,
+		`up -d --wait --wait-timeout 60 postgres redis`,
+		`PROXY_POOL_TEST_POSTGRES_URL`,
+		`PROXY_POOL_TEST_REDIS_URL`,
+		`go test -count=1 -tags=integration -timeout 60s ./internal/controller/bootstrap`,
+		`down --volumes --remove-orphans`,
+	} {
+		if !strings.Contains(script, required) {
+			t.Errorf("test-controller.ps1 missing %q", required)
+		}
+	}
+}
+
 func TestLocalRedisIsExplicitlyEphemeral(t *testing.T) {
 	document := loadComposeDocument(t)
 	redis, ok := document.Services["redis"]

internal/controller/admin/service.go

@@ -178,6 +178,23 @@ func (service *ApplicationService) ReloadConfiguration(ctx context.Context, comm
 			return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err)
 		}
 	}
+	return service.ApplyConfiguration(ctx, command, loaded)
+}
+
+// ApplyConfiguration persists and publishes one already loaded configuration snapshot.
+// Startup and runtime reload paths share this method so storage connections and the
+// committed management view cannot be built from different reads of the source file.
+func (service *ApplicationService) ApplyConfiguration(
+	ctx context.Context,
+	command ReloadCommand,
+	loaded LoadedConfiguration,
+) (MutationResult, error) {
+	if ctx == nil {
+		return MutationResult{RequestID: command.RequestID}, ErrInvalidConfiguration
+	}
+	if err := ctx.Err(); err != nil {
+		return MutationResult{RequestID: command.RequestID}, err
+	}
 	if loaded.Value == nil || strings.TrimSpace(loaded.Source) != loaded.Source || loaded.Source == "" ||
 		len(loaded.Source) > adminstate.MaxSourceBytes {
 		return MutationResult{RequestID: command.RequestID}, ErrInvalidConfiguration

internal/controller/admin/service_test.go

@@ -293,6 +293,42 @@ func TestApplicationServiceReloadPersistsManagementViewBeforePublishing(t *testi
 	}
 }
 
+func TestApplicationServiceApplyConfigurationUsesProvidedSnapshotWithoutReloading(t *testing.T) {
+	t.Parallel()
+	now := time.Date(2026, 7, 30, 9, 0, 0, 0, time.UTC)
+	configuration := validReloadConfiguration()
+	publisher := &recordingConfigurationPublisher{}
+	state := &recordingAdminState{
+		mutation: adminstate.MutationResult{RequestID: "controller-startup", Changed: true, Revision: 1},
+	}
+	service, err := NewApplicationService(ApplicationDependencies{
+		State:         state,
+		Operations:    staticOperationalStatusReader{},
+		Configuration: forbiddenConfigurationLoader{},
+		Publisher:     publisher,
+	}, ApplicationOptions{Now: func() time.Time { return now }})
+	if err != nil {
+		t.Fatalf("NewApplicationService() error = %v", err)
+	}
+
+	result, err := service.ApplyConfiguration(context.Background(), ReloadCommand{
+		RequestID: "controller-startup", ActorID: "proxy-controller",
+	}, LoadedConfiguration{Value: configuration, Source: "configs/controller.yaml"})
+	if err != nil {
+		t.Fatalf("ApplyConfiguration() error = %v", err)
+	}
+	if result.Version != 1 || !result.Changed {
+		t.Fatalf("ApplyConfiguration() result = %+v", result)
+	}
+	if len(publisher.published) != 1 || publisher.published[0] != configuration {
+		t.Fatalf("published configurations = %+v", publisher.published)
+	}
+	if state.lastConfig.Source != "configs/controller.yaml" || state.lastConfig.Actor.ID != "proxy-controller" ||
+		!state.lastConfig.OccurredAt.Equal(now) {
+		t.Fatalf("CommitConfig() metadata = %+v", state.lastConfig)
+	}
+}
+
 func TestApplicationServiceReloadDoesNotPublishInvalidOrUncommittedConfiguration(t *testing.T) {
 	t.Parallel()
 	tests := []struct {
@@ -500,6 +536,12 @@ type staticConfigurationLoader struct {
 	err    error
 }
 
+type forbiddenConfigurationLoader struct{}
+
+func (forbiddenConfigurationLoader) LoadConfiguration(context.Context) (LoadedConfiguration, error) {
+	panic("ApplyConfiguration must not reload the source")
+}
+
 func (loader staticConfigurationLoader) LoadConfiguration(context.Context) (LoadedConfiguration, error) {
 	return loader.loaded, loader.err
 }

internal/controller/bootstrap/bootstrap.go

@@ -0,0 +1,176 @@
+package bootstrap
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"reflect"
+	"strings"
+	"time"
+
+	"proxy-pool/internal/config"
+	"proxy-pool/internal/controller/admin"
+	"proxy-pool/internal/controller/distribution"
+	"proxy-pool/internal/controller/extraction"
+	"proxy-pool/internal/controller/operations"
+	controllerRuntime "proxy-pool/internal/controller/runtime"
+	"proxy-pool/internal/domain/activitypool"
+	extractionDomain "proxy-pool/internal/domain/extraction"
+	"proxy-pool/internal/platform/admission"
+	"proxy-pool/internal/platform/httpserver"
+)
+
+var (
+	ErrInvalidOptions = errors.New("invalid controller bootstrap options")
+	ErrStartup        = errors.New("controller startup failed")
+)
+
+type Options struct {
+	ConfigPath string
+	Resolver   config.Resolver
+	Now        func() time.Time
+	HTTP       httpserver.Options
+}
+
+type activityStore interface {
+	extractionDomain.Store
+	activitypool.StateInventoryReader
+}
+
+type ports struct {
+	state     admin.StateRepository
+	activity  activityStore
+	readiness distribution.ReadinessChecker
+	close     func() error
+}
+
+type infrastructure interface {
+	Open(context.Context, *config.Config) (ports, error)
+}
+
+type controllerRunner interface {
+	Run(context.Context) error
+}
+
+type runtimeFactory interface {
+	New(*config.Config, controllerRuntime.Dependencies, controllerRuntime.Options) (controllerRunner, error)
+}
+
+func Run(ctx context.Context, options Options) error {
+	return run(ctx, options, &productionInfrastructure{}, productionRuntimeFactory{})
+}
+
+func run(ctx context.Context, options Options, infrastructure infrastructure, factory runtimeFactory) (resultErr error) {
+	if ctx == nil || strings.TrimSpace(options.ConfigPath) != options.ConfigPath || options.ConfigPath == "" ||
+		nilInterface(options.Resolver) || nilInterface(infrastructure) || nilInterface(factory) {
+		return ErrInvalidOptions
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	if options.Now == nil {
+		options.Now = time.Now
+	}
+
+	loader, err := admin.NewFileConfigurationLoader(options.ConfigPath, options.Resolver)
+	if err != nil {
+		return errors.Join(ErrInvalidOptions, err)
+	}
+	loaded, err := loader.LoadConfiguration(ctx)
+	if err != nil {
+		return fmt.Errorf("%w: load configuration: %w", ErrStartup, err)
+	}
+	configurationStore, err := config.NewStore(loaded.Value)
+	if err != nil {
+		return fmt.Errorf("%w: initialize configuration store: %w", ErrStartup, err)
+	}
+
+	opened, err := infrastructure.Open(ctx, loaded.Value)
+	if err != nil {
+		return fmt.Errorf("%w: open infrastructure: %w", ErrStartup, err)
+	}
+	if opened.close == nil {
+		return errors.Join(ErrStartup, ErrInvalidOptions)
+	}
+	defer func() {
+		resultErr = errors.Join(resultErr, opened.close())
+	}()
+
+	dependencies := controllerRuntime.Dependencies{}
+	if loaded.Value.Distribution.Enabled {
+		if nilInterface(opened.activity) || nilInterface(opened.readiness) {
+			return errors.Join(ErrStartup, ErrInvalidOptions)
+		}
+		service, serviceErr := extraction.NewService(opened.activity, extractionPolicy(loaded.Value), admission.AllowAll{}, options.Now)
+		if serviceErr != nil {
+			return fmt.Errorf("%w: build extraction service: %w", ErrStartup, serviceErr)
+		}
+		dependencies.Extractor = service
+		dependencies.Readiness = opened.readiness
+	}
+	if loaded.Value.Admin.Enabled {
+		if nilInterface(opened.state) || nilInterface(opened.activity) {
+			return errors.Join(ErrStartup, ErrInvalidOptions)
+		}
+		statusReader, statusErr := operations.NewReader(configurationStore, opened.activity, options.Now)
+		if statusErr != nil {
+			return fmt.Errorf("%w: build operational status reader: %w", ErrStartup, statusErr)
+		}
+		service, serviceErr := admin.NewApplicationService(admin.ApplicationDependencies{
+			State: opened.state, Operations: statusReader, Configuration: loader, Publisher: configurationStore,
+		}, admin.ApplicationOptions{Now: options.Now})
+		if serviceErr != nil {
+			return fmt.Errorf("%w: build admin service: %w", ErrStartup, serviceErr)
+		}
+		if _, applyErr := service.ApplyConfiguration(ctx, admin.ReloadCommand{
+			RequestID: "controller-startup", ActorID: "proxy-controller",
+		}, loaded); applyErr != nil {
+			return fmt.Errorf("%w: commit startup configuration: %w", ErrStartup, applyErr)
+		}
+		dependencies.AdminService = service
+	}
+
+	runner, err := factory.New(configurationStore.Current(), dependencies, controllerRuntime.Options{HTTP: options.HTTP})
+	if err != nil {
+		return fmt.Errorf("%w: build HTTP runtime: %w", ErrStartup, err)
+	}
+	if nilInterface(runner) {
+		return errors.Join(ErrStartup, ErrInvalidOptions)
+	}
+	return runner.Run(ctx)
+}
+
+func extractionPolicy(configuration *config.Config) extraction.Policy {
+	configured := configuration.Distribution.Extraction
+	return extraction.Policy{
+		MaxCountPerRequest: configured.MaxCountPerRequest,
+		DefaultFulfillment: extractionDomain.Fulfillment(configured.Fulfillment),
+		MinRemainingTTL:    configured.MinRemainingTTL.Value(),
+		MaxHealthCheckAge:  configured.MaxHealthCheckAge.Value(),
+		ReserveForGateway:  configured.ReserveForGateway,
+		IdempotencyTTL:     configured.IdempotencyTTL.Value(),
+	}
+}
+
+func nilInterface(value any) bool {
+	if value == nil {
+		return true
+	}
+	reflected := reflect.ValueOf(value)
+	switch reflected.Kind() {
+	case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice:
+		return reflected.IsNil()
+	default:
+		return false
+	}
+}
+
+type productionRuntimeFactory struct{}
+
+func (productionRuntimeFactory) New(
+	configuration *config.Config,
+	dependencies controllerRuntime.Dependencies,
+	options controllerRuntime.Options,
+) (controllerRunner, error) {
+	return controllerRuntime.New(configuration, dependencies, options)
+}

internal/controller/bootstrap/bootstrap_integration_test.go

@@ -0,0 +1,68 @@
+//go:build integration
+
+package bootstrap
+
+import (
+	"context"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"proxy-pool/internal/config"
+	"proxy-pool/internal/controller/admin"
+	controllerRuntime "proxy-pool/internal/controller/runtime"
+)
+
+func TestProductionBootstrapOpensBothStoresAndCommitsStartupConfiguration(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")
+	}
+	source := strings.ReplaceAll(bootstrapTestConfig, "postgres://fixture", postgresURL)
+	source = strings.ReplaceAll(source, "redis://fixture", redisURL)
+	resolver := &memoryResolver{files: map[string][]byte{"controller.yaml": []byte(source)}}
+	factory := &integrationRuntimeFactory{}
+
+	err := run(context.Background(), Options{
+		ConfigPath: "controller.yaml", Resolver: resolver, Now: func() time.Time {
+			return time.Date(2026, 7, 30, 13, 0, 0, 0, time.UTC)
+		},
+	}, &productionInfrastructure{}, factory)
+	if err != nil {
+		t.Fatalf("run() error = %v", err)
+	}
+	if factory.status.ConfigVersion == "" || len(factory.status.Upstreams) != 2 {
+		t.Fatalf("Admin Status = %+v", factory.status)
+	}
+	if factory.status.Upstreams[0].Name != "provider-a" || factory.status.Upstreams[0].Available != 0 ||
+		factory.status.Upstreams[1].Name != "provider-b" {
+		t.Fatalf("Admin Status upstreams = %+v", factory.status.Upstreams)
+	}
+}
+
+type integrationRuntimeFactory struct {
+	status admin.Status
+}
+
+func (factory *integrationRuntimeFactory) New(
+	_ *config.Config,
+	dependencies controllerRuntime.Dependencies,
+	_ controllerRuntime.Options,
+) (controllerRunner, error) {
+	return integrationRunner{run: func(ctx context.Context) error {
+		if err := dependencies.Readiness.Ready(ctx); err != nil {
+			return err
+		}
+		status, err := dependencies.AdminService.Status(ctx)
+		factory.status = status
+		return err
+	}}, nil
+}
+
+type integrationRunner struct {
+	run func(context.Context) error
+}
+
+func (runner integrationRunner) Run(ctx context.Context) error { return runner.run(ctx) }

internal/controller/bootstrap/bootstrap_test.go

@@ -0,0 +1,206 @@
+package bootstrap
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"proxy-pool/internal/config"
+	controllerRuntime "proxy-pool/internal/controller/runtime"
+	"proxy-pool/internal/domain/activitypool"
+	"proxy-pool/internal/domain/adminstate"
+	extractionDomain "proxy-pool/internal/domain/extraction"
+)
+
+func TestRunLoadsOneSnapshotCommitsItAndClosesInfrastructure(t *testing.T) {
+	t.Parallel()
+	resolver := &memoryResolver{files: map[string][]byte{"controller.yaml": []byte(bootstrapTestConfig)}}
+	state := adminstate.NewMemoryStore()
+	activity := &stubActivityStore{}
+	closeErr := errors.New("close failed")
+	infrastructure := &stubInfrastructure{ports: ports{
+		state: state, activity: activity, readiness: readyStub{},
+		close: func() error { return closeErr },
+	}}
+	runErr := errors.New("runtime failed")
+	factory := &recordingRuntimeFactory{runner: runnerStub{err: runErr}}
+	now := time.Date(2026, 7, 30, 11, 0, 0, 0, time.UTC)
+
+	err := run(context.Background(), Options{
+		ConfigPath: "controller.yaml", Resolver: resolver, Now: func() time.Time { return now },
+	}, infrastructure, factory)
+	if !errors.Is(err, runErr) || !errors.Is(err, closeErr) {
+		t.Fatalf("run() error = %v, want runtime and close errors", err)
+	}
+	if resolver.reads != 1 {
+		t.Fatalf("configuration reads = %d, want 1", resolver.reads)
+	}
+	if infrastructure.opens != 1 || infrastructure.configuration == nil {
+		t.Fatalf("infrastructure opens = %d, config = %p", infrastructure.opens, infrastructure.configuration)
+	}
+	snapshot, snapshotErr := state.Snapshot(context.Background())
+	if snapshotErr != nil || snapshot.Config == nil || snapshot.Config.Source != "controller.yaml" || snapshot.Revision != 1 {
+		t.Fatalf("management snapshot = %+v, %v", snapshot, snapshotErr)
+	}
+	if factory.configuration == nil || factory.dependencies.Extractor == nil ||
+		factory.dependencies.Readiness == nil || factory.dependencies.AdminService == nil {
+		t.Fatalf("runtime assembly = config:%p dependencies:%+v", factory.configuration, factory.dependencies)
+	}
+}
+
+func TestRunRejectsInvalidOptionsBeforeIO(t *testing.T) {
+	t.Parallel()
+	valid := Options{ConfigPath: "controller.yaml", Resolver: &memoryResolver{}, Now: time.Now}
+	tests := []struct {
+		name    string
+		ctx     context.Context
+		options Options
+	}{
+		{name: "nil context", options: valid},
+		{name: "missing path", ctx: context.Background(), options: Options{Resolver: valid.Resolver, Now: time.Now}},
+		{name: "unclean path", ctx: context.Background(), options: Options{ConfigPath: " controller.yaml", Resolver: valid.Resolver, Now: time.Now}},
+		{name: "missing resolver", ctx: context.Background(), options: Options{ConfigPath: "controller.yaml", Now: time.Now}},
+	}
+	for _, test := range tests {
+		test := test
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+			if err := run(test.ctx, test.options, &stubInfrastructure{}, &recordingRuntimeFactory{}); !errors.Is(err, ErrInvalidOptions) {
+				t.Fatalf("run() error = %v", err)
+			}
+		})
+	}
+}
+
+type memoryResolver struct {
+	files map[string][]byte
+	reads int
+}
+
+func (*memoryResolver) LookupEnv(string) (string, bool) { return "", false }
+
+func (resolver *memoryResolver) ReadFile(path string) ([]byte, error) {
+	resolver.reads++
+	content, ok := resolver.files[path]
+	if !ok {
+		return nil, errors.New("file missing")
+	}
+	return append([]byte(nil), content...), nil
+}
+
+type stubInfrastructure struct {
+	ports         ports
+	err           error
+	opens         int
+	configuration *config.Config
+}
+
+func (infrastructure *stubInfrastructure) Open(
+	_ context.Context,
+	configuration *config.Config,
+) (ports, error) {
+	infrastructure.opens++
+	infrastructure.configuration = configuration
+	return infrastructure.ports, infrastructure.err
+}
+
+type recordingRuntimeFactory struct {
+	configuration *config.Config
+	dependencies  controllerRuntime.Dependencies
+	runner        controllerRunner
+	err           error
+}
+
+func (factory *recordingRuntimeFactory) New(
+	configuration *config.Config,
+	dependencies controllerRuntime.Dependencies,
+	options controllerRuntime.Options,
+) (controllerRunner, error) {
+	factory.configuration = configuration
+	factory.dependencies = dependencies
+	return factory.runner, factory.err
+}
+
+type runnerStub struct{ err error }
+
+func (runner runnerStub) Run(context.Context) error { return runner.err }
+
+type readyStub struct{}
+
+func (readyStub) Ready(context.Context) error { return nil }
+
+type stubActivityStore struct{}
+
+func (*stubActivityStore) Extract(_ context.Context, command extractionDomain.Command) (extractionDomain.Result, error) {
+	return extractionDomain.Result{Requested: command.Requested}, nil
+}
+
+func (*stubActivityStore) ReadStateInventory(
+	_ context.Context,
+	upstreamIDs []string,
+	_ time.Time,
+) ([]activitypool.StateInventory, error) {
+	result := make([]activitypool.StateInventory, len(upstreamIDs))
+	for index, upstreamID := range upstreamIDs {
+		result[index].UpstreamID = upstreamID
+	}
+	return result, nil
+}
+
+const bootstrapTestConfig = `
+version: 1
+security:
+  requireProtectionOnPublicListen: true
+gateway:
+  enabled: false
+distribution:
+  enabled: true
+  listen: 127.0.0.1:0
+  auth: {mode: none}
+  extraction:
+    fulfillment: partial
+    maxCountPerRequest: 20
+    minRemainingTTL: 5s
+    maxHealthCheckAge: 15s
+    reserveForGateway: 5
+    idempotencyTTL: 5m
+admin:
+  enabled: true
+  listen: 127.0.0.1:0
+  auth: {mode: none}
+storage:
+  postgresURL: postgres://fixture
+  redisURL: redis://fixture
+routing:
+  - name: extract
+    enabled: true
+    purpose: extract
+    upstreams: [provider-a, provider-b]
+    strategy: {type: sequential, switchAfterEmptyFetch: 5, endBehavior: stayLast}
+    onUnavailable: {action: reject}
+upstreams:
+  provider-a: &upstream
+    enabled: true
+    exposure: [extract]
+    provider: {billingMode: fetch, protocols: [http]}
+    api:
+      url: https://provider.invalid/proxies
+      method: GET
+      template: '{{.}}'
+      auth: {type: none}
+    proxyAuth: {type: response}
+    pool: {maxSize: 100}
+    capacity: {maxConcurrencyPerProxy: 10}
+    lifecycle: {ttl: 2m, allocationSafetyMargin: 10s}
+    fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1, maxTotal: 1000}
+    check:
+      interval: 30s
+      jitter: 20
+      maxInFlight: 100
+      timeout: 2s
+      maxAttempts: 2
+      maxConsecutiveFailures: 3
+      urls: [https://example.invalid/health]
+  provider-b: *upstream
+`

internal/controller/bootstrap/infrastructure.go

@@ -0,0 +1,166 @@
+package bootstrap
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/redis/go-redis/v9"
+
+	"proxy-pool/internal/adapters/postgresadmin"
+	"proxy-pool/internal/adapters/redisactivity"
+	"proxy-pool/internal/config"
+	"proxy-pool/internal/platform/credentials"
+)
+
+const (
+	redisNamespace    = "controller"
+	redisOperationTTL = 30 * time.Second
+	redisMinimumScan  = 4_096
+	redisCleanupLimit = 1_024
+)
+
+var (
+	ErrPostgresConfiguration = errors.New("invalid PostgreSQL configuration")
+	ErrPostgresUnavailable   = errors.New("PostgreSQL unavailable")
+	ErrRedisConfiguration    = errors.New("invalid Redis configuration")
+	ErrRedisUnavailable      = errors.New("Redis unavailable")
+)
+
+type productionInfrastructure struct{}
+
+func (*productionInfrastructure) Open(
+	ctx context.Context,
+	configuration *config.Config,
+) (_ ports, resultErr error) {
+	if ctx == nil || configuration == nil {
+		return ports{}, ErrInvalidOptions
+	}
+	var postgresPool *pgxpool.Pool
+	var redisClient *redis.Client
+	closeResources := func() error {
+		var closeErr error
+		if redisClient != nil {
+			closeErr = redisClient.Close()
+		}
+		if postgresPool != nil {
+			postgresPool.Close()
+		}
+		return closeErr
+	}
+	defer func() {
+		if resultErr != nil {
+			_ = closeResources()
+		}
+	}()
+
+	opened := ports{close: closeResources}
+	if configuration.Admin.Enabled {
+		if strings.TrimSpace(configuration.Storage.PostgresURL) == "" {
+			return ports{}, ErrPostgresConfiguration
+		}
+		poolConfig, err := pgxpool.ParseConfig(configuration.Storage.PostgresURL)
+		if err != nil {
+			return ports{}, ErrPostgresConfiguration
+		}
+		poolConfig.ConnConfig.RuntimeParams["application_name"] = "proxy-controller"
+		postgresPool, err = pgxpool.NewWithConfig(ctx, poolConfig)
+		if err != nil {
+			return ports{}, ErrPostgresUnavailable
+		}
+		if err = postgresPool.Ping(ctx); err != nil {
+			return ports{}, contextOr(ctx, ErrPostgresUnavailable)
+		}
+		if err = postgresadmin.ApplyMigrations(ctx, postgresPool); err != nil {
+			return ports{}, err
+		}
+		opened.state, err = postgresadmin.New(postgresPool)
+		if err != nil {
+			return ports{}, err
+		}
+	}
+
+	if configuration.Distribution.Enabled || configuration.Admin.Enabled {
+		if strings.TrimSpace(configuration.Storage.RedisURL) == "" {
+			return ports{}, ErrRedisConfiguration
+		}
+		redisOptions, err := redis.ParseURL(configuration.Storage.RedisURL)
+		if err != nil {
+			return ports{}, ErrRedisConfiguration
+		}
+		redisClient = redis.NewClient(redisOptions)
+		if err = redisClient.Ping(ctx).Err(); err != nil {
+			return ports{}, contextOr(ctx, ErrRedisUnavailable)
+		}
+		credentialStore, err := credentials.NewMemoryStore(credentialCapacity(configuration))
+		if err != nil {
+			return ports{}, err
+		}
+		adapter, err := redisactivity.New(redisClient, redisactivity.Options{
+			Namespace:        redisNamespace,
+			Credentials:      credentialStore,
+			OperationTTL:     redisOperationTTL,
+			MaxCandidateScan: candidateScan(configuration),
+			CleanupLimit:     redisCleanupLimit,
+		})
+		if err != nil {
+			return ports{}, err
+		}
+		opened.activity = adapter
+		opened.readiness = redisReadiness{client: redisClient}
+	}
+	return opened, nil
+}
+
+type redisReadiness struct {
+	client *redis.Client
+}
+
+func (readiness redisReadiness) Ready(ctx context.Context) error {
+	if ctx == nil || readiness.client == nil {
+		return ErrRedisUnavailable
+	}
+	if err := readiness.client.Ping(ctx).Err(); err != nil {
+		return contextOr(ctx, ErrRedisUnavailable)
+	}
+	return nil
+}
+
+func credentialCapacity(configuration *config.Config) int {
+	capacity := 0
+	maximum := int(^uint(0) >> 1)
+	for _, upstream := range configuration.Upstreams {
+		if upstream.Pool.MaxSize <= 0 {
+			continue
+		}
+		if capacity > maximum-upstream.Pool.MaxSize {
+			return maximum
+		}
+		capacity += upstream.Pool.MaxSize
+	}
+	if capacity == 0 {
+		return 1
+	}
+	return capacity
+}
+
+func candidateScan(configuration *config.Config) int {
+	configured := configuration.Distribution.Extraction
+	if configured.MaxCountPerRequest > int(^uint(0)>>1)-configured.ReserveForGateway {
+		return int(^uint(0) >> 1)
+	}
+	value := configured.MaxCountPerRequest + configured.ReserveForGateway
+	if value < redisMinimumScan {
+		return redisMinimumScan
+	}
+	return value
+}
+
+func contextOr(ctx context.Context, fallback error) error {
+	if ctx != nil && ctx.Err() != nil {
+		return ctx.Err()
+	}
+	return fallback
+}

internal/controller/bootstrap/infrastructure_test.go

@@ -0,0 +1,54 @@
+package bootstrap
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"proxy-pool/internal/config"
+)
+
+func TestProductionInfrastructureRejectsInvalidStorageWithoutLeakingURLs(t *testing.T) {
+	t.Parallel()
+	postgresSecret := "postgres-secret"
+	_, err := (&productionInfrastructure{}).Open(context.Background(), &config.Config{
+		Admin:   config.Listener{Enabled: true},
+		Storage: config.Storage{PostgresURL: "postgres://user:" + postgresSecret + "@%zz"},
+	})
+	if !errors.Is(err, ErrPostgresConfiguration) || strings.Contains(err.Error(), postgresSecret) {
+		t.Fatalf("Open(invalid PostgreSQL) error = %v", err)
+	}
+
+	redisSecret := "redis-secret"
+	_, err = (&productionInfrastructure{}).Open(context.Background(), &config.Config{
+		Distribution: config.Distribution{Listener: config.Listener{Enabled: true}},
+		Storage:      config.Storage{RedisURL: "redis://user:" + redisSecret + "@%zz"},
+	})
+	if !errors.Is(err, ErrRedisConfiguration) || strings.Contains(err.Error(), redisSecret) {
+		t.Fatalf("Open(invalid Redis) error = %v", err)
+	}
+}
+
+func TestProductionRedisSizingUsesConfigurationBounds(t *testing.T) {
+	t.Parallel()
+	configuration := &config.Config{
+		Distribution: config.Distribution{Extraction: config.Extraction{
+			MaxCountPerRequest: 100, ReserveForGateway: 5_000,
+		}},
+		Upstreams: map[string]config.Upstream{
+			"provider-a": {Pool: config.Pool{MaxSize: 3_000}},
+			"provider-b": {Pool: config.Pool{MaxSize: 2_000}},
+		},
+	}
+	if got := credentialCapacity(configuration); got != 5_000 {
+		t.Fatalf("credentialCapacity() = %d, want 5000", got)
+	}
+	if got := candidateScan(configuration); got != 5_100 {
+		t.Fatalf("candidateScan() = %d, want 5100", got)
+	}
+	configuration.Distribution.Extraction = config.Extraction{MaxCountPerRequest: 1}
+	if got := candidateScan(configuration); got != redisMinimumScan {
+		t.Fatalf("candidateScan(minimum) = %d, want %d", got, redisMinimumScan)
+	}
+}

internal/controller/operations/reader.go

@@ -0,0 +1,104 @@
+package operations
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"sort"
+	"time"
+
+	"proxy-pool/internal/config"
+	"proxy-pool/internal/controller/admin"
+	"proxy-pool/internal/domain/activitypool"
+)
+
+var (
+	ErrInvalidReader = errors.New("invalid operational status reader")
+	ErrUnavailable   = errors.New("operational status unavailable")
+)
+
+type ConfigurationReader interface {
+	Current() *config.Config
+}
+
+type Reader struct {
+	configuration ConfigurationReader
+	inventory     activitypool.StateInventoryReader
+	now           func() time.Time
+}
+
+var _ admin.OperationalStatusReader = (*Reader)(nil)
+
+func NewReader(
+	configuration ConfigurationReader,
+	inventory activitypool.StateInventoryReader,
+	now func() time.Time,
+) (*Reader, error) {
+	if nilInterface(configuration) || nilInterface(inventory) || now == nil {
+		return nil, ErrInvalidReader
+	}
+	return &Reader{configuration: configuration, inventory: inventory, now: now}, nil
+}
+
+func (reader *Reader) ReadOperationalStatus(ctx context.Context) (admin.OperationalStatus, error) {
+	if ctx == nil || reader == nil {
+		return admin.OperationalStatus{}, ErrInvalidReader
+	}
+	if err := ctx.Err(); err != nil {
+		return admin.OperationalStatus{}, err
+	}
+	configuration := reader.configuration.Current()
+	if configuration == nil {
+		return admin.OperationalStatus{}, ErrUnavailable
+	}
+
+	upstreamIDs := make([]string, 0, len(configuration.Upstreams))
+	for upstreamID := range configuration.Upstreams {
+		upstreamIDs = append(upstreamIDs, upstreamID)
+	}
+	sort.Strings(upstreamIDs)
+	inventories, err := reader.inventory.ReadStateInventory(ctx, upstreamIDs, reader.now().UTC())
+	if err != nil {
+		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+			return admin.OperationalStatus{}, err
+		}
+		return admin.OperationalStatus{}, errors.Join(ErrUnavailable, err)
+	}
+	if len(inventories) != len(upstreamIDs) {
+		return admin.OperationalStatus{}, ErrUnavailable
+	}
+
+	status := admin.OperationalStatus{Upstreams: make([]admin.UpstreamActivity, len(inventories))}
+	for index, inventory := range inventories {
+		if inventory.UpstreamID != upstreamIDs[index] || invalidInventory(inventory) {
+			return admin.OperationalStatus{}, ErrUnavailable
+		}
+		status.Upstreams[index] = admin.UpstreamActivity{
+			Name:      inventory.UpstreamID,
+			Available: inventory.Available,
+			Checking:  inventory.Checking,
+			Suspect:   inventory.Suspect,
+			Draining:  inventory.Draining,
+			Extracted: inventory.Extracted,
+		}
+	}
+	return status, nil
+}
+
+func invalidInventory(inventory activitypool.StateInventory) bool {
+	return inventory.Fetched < 0 || inventory.Checking < 0 || inventory.Available < 0 ||
+		inventory.Suspect < 0 || inventory.Draining < 0 || inventory.Unhealthy < 0 || inventory.Extracted < 0
+}
+
+func nilInterface(value any) bool {
+	if value == nil {
+		return true
+	}
+	reflected := reflect.ValueOf(value)
+	switch reflected.Kind() {
+	case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice:
+		return reflected.IsNil()
+	default:
+		return false
+	}
+}

internal/controller/operations/reader_test.go

@@ -0,0 +1,118 @@
+package operations
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"proxy-pool/internal/config"
+	"proxy-pool/internal/domain/activitypool"
+)
+
+func TestReaderMapsCurrentUpstreamsToAdminOperationalStatus(t *testing.T) {
+	t.Parallel()
+	now := time.Date(2026, 7, 30, 10, 0, 0, 0, time.UTC)
+	inventory := &recordingStateInventoryReader{result: []activitypool.StateInventory{
+		{UpstreamID: "provider-a", Fetched: 3, Checking: 2, Available: 11, Suspect: 1, Draining: 4, Extracted: 8},
+		{UpstreamID: "provider-b", Available: 7},
+	}}
+	reader, err := NewReader(staticConfigurationReader{configuration: &config.Config{
+		Upstreams: map[string]config.Upstream{"provider-b": {}, "provider-a": {}},
+	}}, inventory, func() time.Time { return now })
+	if err != nil {
+		t.Fatalf("NewReader() error = %v", err)
+	}
+
+	status, err := reader.ReadOperationalStatus(context.Background())
+	if err != nil {
+		t.Fatalf("ReadOperationalStatus() error = %v", err)
+	}
+	if len(inventory.upstreamIDs) != 2 || inventory.upstreamIDs[0] != "provider-a" || inventory.upstreamIDs[1] != "provider-b" ||
+		!inventory.now.Equal(now) {
+		t.Fatalf("ReadStateInventory() input = %+v at %v", inventory.upstreamIDs, inventory.now)
+	}
+	if status.SnapshotVersion != 0 || len(status.Workers) != 0 || len(status.Upstreams) != 2 {
+		t.Fatalf("operational status shape = %+v", status)
+	}
+	first := status.Upstreams[0]
+	if first.Name != "provider-a" || first.Available != 11 || first.Checking != 2 || first.Suspect != 1 ||
+		first.Draining != 4 || first.Extracted != 8 {
+		t.Fatalf("first upstream = %+v", first)
+	}
+	if status.Upstreams[1].Name != "provider-b" || status.Upstreams[1].Available != 7 {
+		t.Fatalf("second upstream = %+v", status.Upstreams[1])
+	}
+}
+
+func TestReaderRejectsMissingOrMalformedDependencies(t *testing.T) {
+	t.Parallel()
+	now := func() time.Time { return time.Now().UTC() }
+	validConfig := staticConfigurationReader{configuration: &config.Config{
+		Upstreams: map[string]config.Upstream{"provider-a": {}},
+	}}
+	validInventory := &recordingStateInventoryReader{result: []activitypool.StateInventory{{UpstreamID: "provider-a"}}}
+
+	if _, err := NewReader(nil, validInventory, now); !errors.Is(err, ErrInvalidReader) {
+		t.Fatalf("NewReader(nil config) error = %v", err)
+	}
+	if _, err := NewReader(validConfig, nil, now); !errors.Is(err, ErrInvalidReader) {
+		t.Fatalf("NewReader(nil inventory) error = %v", err)
+	}
+	if _, err := NewReader(validConfig, validInventory, nil); !errors.Is(err, ErrInvalidReader) {
+		t.Fatalf("NewReader(nil clock) error = %v", err)
+	}
+
+	malformed := &recordingStateInventoryReader{result: []activitypool.StateInventory{{UpstreamID: "wrong"}}}
+	reader, err := NewReader(validConfig, malformed, now)
+	if err != nil {
+		t.Fatalf("NewReader() error = %v", err)
+	}
+	if _, err := reader.ReadOperationalStatus(context.Background()); !errors.Is(err, ErrUnavailable) {
+		t.Fatalf("ReadOperationalStatus(malformed) error = %v", err)
+	}
+}
+
+func TestReaderPreservesCancellationAndClassifiesDependencyFailures(t *testing.T) {
+	t.Parallel()
+	dependencyError := errors.New("redis down")
+	inventory := &recordingStateInventoryReader{err: dependencyError}
+	reader, err := NewReader(staticConfigurationReader{configuration: &config.Config{
+		Upstreams: map[string]config.Upstream{"provider-a": {}},
+	}}, inventory, time.Now)
+	if err != nil {
+		t.Fatalf("NewReader() error = %v", err)
+	}
+	if _, err := reader.ReadOperationalStatus(context.Background()); !errors.Is(err, ErrUnavailable) || !errors.Is(err, dependencyError) {
+		t.Fatalf("ReadOperationalStatus(dependency) error = %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if _, err := reader.ReadOperationalStatus(ctx); !errors.Is(err, context.Canceled) || errors.Is(err, ErrUnavailable) {
+		t.Fatalf("ReadOperationalStatus(canceled) error = %v", err)
+	}
+}
+
+type staticConfigurationReader struct {
+	configuration *config.Config
+}
+
+func (reader staticConfigurationReader) Current() *config.Config { return reader.configuration }
+
+type recordingStateInventoryReader struct {
+	result      []activitypool.StateInventory
+	err         error
+	upstreamIDs []string
+	now         time.Time
+}
+
+func (reader *recordingStateInventoryReader) ReadStateInventory(
+	_ context.Context,
+	upstreamIDs []string,
+	now time.Time,
+) ([]activitypool.StateInventory, error) {
+	reader.upstreamIDs = append([]string(nil), upstreamIDs...)
+	reader.now = now
+	return append([]activitypool.StateInventory(nil), reader.result...), reader.err
+}

internal/platform/admission/allow_all.go

@@ -0,0 +1,14 @@
+package admission
+
+import "context"
+
+// AllowAll validates the common admission contract without applying another
+// quota. It is used when an outer transport boundary already owns rate limits.
+type AllowAll struct{}
+
+func (AllowAll) Admit(ctx context.Context, key string) error {
+	if ctx == nil || key == "" {
+		return ErrInvalidIdentity
+	}
+	return ctx.Err()
+}

internal/platform/admission/allow_all_test.go

@@ -0,0 +1,23 @@
+package admission
+
+import (
+	"context"
+	"errors"
+	"testing"
+)
+
+func TestAllowAllPreservesContextAndIdentityValidation(t *testing.T) {
+	t.Parallel()
+	var admission AllowAll
+	if err := admission.Admit(context.Background(), "client-a"); err != nil {
+		t.Fatalf("Admit(valid) error = %v", err)
+	}
+	if err := admission.Admit(context.Background(), ""); !errors.Is(err, ErrInvalidIdentity) {
+		t.Fatalf("Admit(empty identity) error = %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if err := admission.Admit(ctx, "client-a"); !errors.Is(err, context.Canceled) {
+		t.Fatalf("Admit(canceled) error = %v", err)
+	}
+}

scripts/test-controller.ps1

@@ -0,0 +1,42 @@
+$ErrorActionPreference = "Stop"
+
+$repositoryRoot = Split-Path -Parent $PSScriptRoot
+$composeFile = Join-Path $repositoryRoot "deploy/docker-compose.test.yml"
+$composeProject = "proxy-pool-controller-test"
+$previousPostgresURL = [Environment]::GetEnvironmentVariable("PROXY_POOL_TEST_POSTGRES_URL", "Process")
+$previousRedisURL = [Environment]::GetEnvironmentVariable("PROXY_POOL_TEST_REDIS_URL", "Process")
+
+try {
+    docker compose -p $composeProject -f $composeFile up -d --wait --wait-timeout 60 postgres redis
+    if ($LASTEXITCODE -ne 0) {
+        throw "starting Controller test fixtures failed with exit code $LASTEXITCODE"
+    }
+
+    $env:PROXY_POOL_TEST_POSTGRES_URL = "postgres://proxy_pool_test:proxy-pool-test@127.0.0.1:15432/proxy_pool_test?sslmode=disable"
+    $env:PROXY_POOL_TEST_REDIS_URL = "redis://127.0.0.1:16379/15"
+    Push-Location $repositoryRoot
+    try {
+        go test -count=1 -tags=integration -timeout 60s ./internal/controller/bootstrap
+        if ($LASTEXITCODE -ne 0) {
+            throw "Controller integration tests failed with exit code $LASTEXITCODE"
+        }
+    }
+    finally {
+        Pop-Location
+    }
+}
+finally {
+    if ($null -eq $previousPostgresURL) {
+        Remove-Item Env:PROXY_POOL_TEST_POSTGRES_URL -ErrorAction SilentlyContinue
+    }
+    else {
+        $env:PROXY_POOL_TEST_POSTGRES_URL = $previousPostgresURL
+    }
+    if ($null -eq $previousRedisURL) {
+        Remove-Item Env:PROXY_POOL_TEST_REDIS_URL -ErrorAction SilentlyContinue
+    }
+    else {
+        $env:PROXY_POOL_TEST_REDIS_URL = $previousRedisURL
+    }
+    docker compose -p $composeProject -f $composeFile down --volumes --remove-orphans
+}