diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c6d4433..b8045aa 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/.gitignore b/.gitignore index 0d61e3d..4d8c67a 100644 --- a/.gitignore +++ b/.gitignore @@ -9,6 +9,7 @@ coverage/ *.out *.test *.prof +*.exe .tmp-proto/ # Local configuration and secrets diff --git a/cmd/proxy-controller/main.go b/cmd/proxy-controller/main.go new file mode 100644 index 0000000..28d79f1 --- /dev/null +++ b/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 +} diff --git a/cmd/proxy-controller/main_test.go b/cmd/proxy-controller/main_test.go new file mode 100644 index 0000000..5581e25 --- /dev/null +++ b/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()) + } +} diff --git a/deploy/compose_test.go b/deploy/compose_test.go index e84cc34..9c8be29 100644 --- a/deploy/compose_test.go +++ b/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"] diff --git a/internal/controller/admin/service.go b/internal/controller/admin/service.go index fc9e2db..eb0706c 100644 --- a/internal/controller/admin/service.go +++ b/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 diff --git a/internal/controller/admin/service_test.go b/internal/controller/admin/service_test.go index ee4c092..94ee622 100644 --- a/internal/controller/admin/service_test.go +++ b/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 } diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go new file mode 100644 index 0000000..2031609 --- /dev/null +++ b/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) +} diff --git a/internal/controller/bootstrap/bootstrap_integration_test.go b/internal/controller/bootstrap/bootstrap_integration_test.go new file mode 100644 index 0000000..f9de8bb --- /dev/null +++ b/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) } diff --git a/internal/controller/bootstrap/bootstrap_test.go b/internal/controller/bootstrap/bootstrap_test.go new file mode 100644 index 0000000..2b11883 --- /dev/null +++ b/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 +` diff --git a/internal/controller/bootstrap/infrastructure.go b/internal/controller/bootstrap/infrastructure.go new file mode 100644 index 0000000..d851dab --- /dev/null +++ b/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 +} diff --git a/internal/controller/bootstrap/infrastructure_test.go b/internal/controller/bootstrap/infrastructure_test.go new file mode 100644 index 0000000..eadff90 --- /dev/null +++ b/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) + } +} diff --git a/internal/controller/operations/reader.go b/internal/controller/operations/reader.go new file mode 100644 index 0000000..3be59fc --- /dev/null +++ b/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 + } +} diff --git a/internal/controller/operations/reader_test.go b/internal/controller/operations/reader_test.go new file mode 100644 index 0000000..493461d --- /dev/null +++ b/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 +} diff --git a/internal/platform/admission/allow_all.go b/internal/platform/admission/allow_all.go new file mode 100644 index 0000000..c3543f8 --- /dev/null +++ b/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() +} diff --git a/internal/platform/admission/allow_all_test.go b/internal/platform/admission/allow_all_test.go new file mode 100644 index 0000000..6ebdcb8 --- /dev/null +++ b/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) + } +} diff --git a/scripts/test-controller.ps1 b/scripts/test-controller.ps1 new file mode 100644 index 0000000..3f68a15 --- /dev/null +++ b/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 +}