diff --git a/docs/api/admin.md b/docs/api/admin.md index 8667033..cc467b9 100644 --- a/docs/api/admin.md +++ b/docs/api/admin.md @@ -33,6 +33,20 @@ Actor ID 与可信 SourceIP 会进入所有 mutation 命令,但认证材料不 `controller/runtime` 将 Admin 与 Distribution 放在不同 `net.Listener`,任一 监听器异常会触发同组端点的有界优雅停机。 +`admin.ApplicationService` 把 Handler DTO 映射到 `adminstate` 公用事务命令。 +Status 以一个权威管理快照决定 Upstream 集合和 Enabled 状态,只从注入的运行态 +读取器补充低基数计数、Worker 与已发布 Snapshot 版本。配置重载顺序固定为: + +1. `FileConfigurationLoader` 通过 `config.LoadResolved` 严格解析、解析 Secret 引用 + 并完成全量校验。 +2. 从脱敏管理投影计算版本与校验和;Secret 值及其可验证摘要不进入管理状态。 +3. 在同一 `adminstate` mutation 中提交配置修订、管理状态、审计和 Outbox。 +4. 提交成功后由 `config.Store` 一次原子指针交换发布完整运行配置;提交失败时旧 + 配置保持不变。幂等重放仍执行发布,以修复进程本地状态。 + +主配置或 Secret 文件 I/O 故障归类为 503;语法、未知字段、引用和语义校验失败 +归类为 422。请求取消和截止时间保持原始上下文错误,不误报为配置错误。 + 除契约中的 401/403/404/409/422 外,运行时还明确返回: - `400`:Request ID 或 JSON 无效。 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 0963c65..475805c 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -184,7 +184,10 @@ Admin/Distribution 必需依赖。共享 `platform/httpserver` 与 `controller/runtime` 已完成 Distribution/Admin 独立监听器、首错联动关闭和 有界优雅停机。Admin mutation 已携带认证 Actor/SourceIP;公用 `adminstate` 事务契约、MemoryStore、100 并发 Routing CAS、租约 Outbox 和六表管理 Schema -已完成。生产命令入口、pgx Adapter 及 PostgreSQL 18 集成测试仍待实现。 +已完成。Admin `ApplicationService` 已将 mutation、权威管理快照、低基数运行态 +聚合与配置重载接到同一公用 seam;严格文件加载、脱敏管理摘要及原子配置发布 +已通过失败路径和并发测试。生产命令入口、pgx Adapter 及 PostgreSQL 18 集成 +测试仍待实现。 已新增公用 `domain/activitypool` 契约及并发安全内存参考实现,Provider Reconciler 通过 `UpsertFetched` 写入带供应商 TTL 和分配安全余量的批次;已覆盖 diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 82c605b..18921e0 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -44,7 +44,9 @@ 运行同一公用契约。 - `PostgreSQL 管理面基础`:已定义 `adminstate` 事务 seam、并发安全 MemoryStore、 公用契约、100 并发 Routing CAS、租约 Outbox 和只含六张管理表的 Schema; - Admin Handler 已向 mutation 传播 Actor/SourceIP。 + Admin Handler 已向 mutation 传播 Actor/SourceIP。Admin ApplicationService 已 + 完成管理 mutation 映射、权威/运行态 Status 聚合、严格配置加载和持久化成功后 + 的原子发布;边界测试禁止其依赖 Redis Extract 或 Proxy 明细包。 ## 2. 已执行验证 diff --git a/docs/superpowers/plans/2026-07-29-postgresql-admin-state.md b/docs/superpowers/plans/2026-07-29-postgresql-admin-state.md index 5ef50e6..5daff49 100644 --- a/docs/superpowers/plans/2026-07-29-postgresql-admin-state.md +++ b/docs/superpowers/plans/2026-07-29-postgresql-admin-state.md @@ -120,15 +120,15 @@ contract; Controller Admin maps its existing typed commands onto this seam. - [x] Change Admin protection to resolve `httpsecurity.Identity` once and add actor/source IP to mutation commands without exposing credentials. -- [ ] Map typed Handler commands to `adminstate.Mutator`; map domain conflict/not-found/ +- [x] Map typed Handler commands to `adminstate.Mutator`; map domain conflict/not-found/ invalid/unavailable errors to the existing HTTP contract. -- [ ] Build Status from one adminstate Snapshot plus injected activity/worker aggregate readers. -- [ ] Reload configuration with existing strict loader/validator, submit a secret-free management +- [x] Build Status from one adminstate Snapshot plus injected activity/worker aggregate readers. +- [x] Reload configuration with existing strict loader/validator, submit a secret-free management snapshot, then atomically publish runtime config only after persistence succeeds. -- [ ] Test persistence failure leaves the current runtime config unchanged and Redis Extract is +- [x] Test persistence failure leaves the current runtime config unchanged and Redis Extract is not referenced by the Admin module. -- [ ] Run `go test -count=1 -timeout 60s ./internal/controller/admin/...`. -- [ ] Commit with `feat: connect admin API to management state`. +- [x] Run `go test -count=1 -timeout 60s ./internal/controller/admin/...`. +- [x] Commit with `feat: connect admin API to management state`. ### Task 7: Documentation and Delivery Verification diff --git a/docs/testing/test-strategy.md b/docs/testing/test-strategy.md index 66c0947..3cd7743 100644 --- a/docs/testing/test-strategy.md +++ b/docs/testing/test-strategy.md @@ -93,6 +93,11 @@ PostgreSQL 管理面使用 `adminstate/contracttest` 作为 Memory/PostgreSQL CAS、审计和租约 Outbox;Schema 静态测试证明只声明六张管理表。只有 pgx Adapter 在真实 PostgreSQL 18 上运行同一契约后,才标记生产持久化完成。 +Admin 应用层测试覆盖 typed-nil 依赖、Actor/SourceIP 映射、Routing CAS 错误、 +权威管理快照与低基数运行态聚合、未知字段拒绝、主配置/Secret 文件 I/O 分类、 +持久化失败不发布、幂等重放发布、脱敏管理摘要和原子配置 Store 并发读写。静态 +导入边界测试禁止 Admin 引用 Redis Activity/Extract 与 Proxy 明细包。 + 需要 PostgreSQL/Redis 的测试使用独立实例和短生命周期容器,不复用开发数据。 测试结束后验证没有残留 Worker ownership、Leader 租约、活动池条目或幂等键, 并检查 PostgreSQL 中不存在 Proxy 明细和逐次提取记录。 diff --git a/internal/config/store.go b/internal/config/store.go new file mode 100644 index 0000000..ca3a460 --- /dev/null +++ b/internal/config/store.go @@ -0,0 +1,43 @@ +package config + +import ( + "errors" + "sync/atomic" +) + +var ErrInvalidStore = errors.New("invalid configuration store") + +// Store publishes complete validated configurations with one atomic pointer swap. +type Store struct { + current atomic.Pointer[Config] +} + +func NewStore(initial *Config) (*Store, error) { + if err := Validate(initial); err != nil { + return nil, errors.Join(ErrInvalidStore, err) + } + store := &Store{} + store.Publish(initial) + return store, nil +} + +func (store *Store) Current() *Config { + if store == nil { + return nil + } + current := store.current.Load() + if current == nil { + return nil + } + cloned := cloneConfig(*current) + return &cloned +} + +// Publish accepts a non-nil configuration already validated by the caller. +func (store *Store) Publish(configuration *Config) { + if store == nil || configuration == nil { + return + } + cloned := cloneConfig(*configuration) + store.current.Store(&cloned) +} diff --git a/internal/config/store_test.go b/internal/config/store_test.go new file mode 100644 index 0000000..45eed28 --- /dev/null +++ b/internal/config/store_test.go @@ -0,0 +1,89 @@ +package config + +import ( + "errors" + "sync" + "testing" +) + +func TestStorePublishesAndReturnsDetachedConfigurations(t *testing.T) { + t.Parallel() + initial := storeTestConfig("provider-a") + store, err := NewStore(initial) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + + initial.Upstreams["provider-a"] = Upstream{} + current := store.Current() + if current == nil || !current.Upstreams["provider-a"].Enabled { + t.Fatalf("Current() was changed through constructor input: %+v", current) + } + current.Routing[0].Upstreams[0] = "mutated" + if got := store.Current().Routing[0].Upstreams[0]; got != "provider-a" { + t.Fatalf("Current() shared mutable state: %q", got) + } + + next := storeTestConfig("provider-b") + store.Publish(next) + next.Routing[0].Upstreams[0] = "mutated" + if got := store.Current().Routing[0].Upstreams[0]; got != "provider-b" { + t.Fatalf("Publish() retained caller state: %q", got) + } +} + +func TestStoreRejectsInvalidInitialConfiguration(t *testing.T) { + t.Parallel() + for _, configuration := range []*Config{nil, {}} { + if _, err := NewStore(configuration); !errors.Is(err, ErrInvalidStore) { + t.Fatalf("NewStore(%v) error = %v, want %v", configuration, err, ErrInvalidStore) + } + } +} + +func TestStoreSupportsConcurrentReadersAndPublishers(t *testing.T) { + store, err := NewStore(storeTestConfig("provider-a")) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + var wait sync.WaitGroup + for index := 0; index < 100; index++ { + wait.Add(2) + go func(index int) { + defer wait.Done() + name := "provider-a" + if index%2 == 1 { + name = "provider-b" + } + store.Publish(storeTestConfig(name)) + }(index) + go func() { + defer wait.Done() + current := store.Current() + if current == nil || len(current.Routing) != 1 || len(current.Routing[0].Upstreams) != 1 { + t.Errorf("Current() returned partial configuration: %+v", current) + } + }() + } + wait.Wait() +} + +func storeTestConfig(upstreamName string) *Config { + return &Config{ + Version: 1, + Upstreams: map[string]Upstream{ + upstreamName: { + Enabled: true, Exposure: []string{"gateway"}, + API: ProviderAPI{Auth: ProviderAuth{Type: "none"}}, + ProxyAuth: ProxyAuth{Type: "response"}, + Pool: Pool{MaxSize: 10}, Capacity: Capacity{MaxConcurrencyPerProxy: 1}, + Lifecycle: Lifecycle{TTL: Duration(60_000_000_000), AllocationSafetyMargin: Duration(10_000_000_000)}, + Fetch: Fetch{Timeout: Duration(1_000_000_000), MaxAttempts: 1, MaxInFlight: 1}, + }, + }, + Routing: []Routing{{ + Name: "default", Enabled: true, Purpose: "gateway", Upstreams: []string{upstreamName}, + Strategy: Strategy{Type: "random"}, OnUnavailable: OnUnavailable{Action: "reject"}, + }}, + } +} diff --git a/internal/controller/admin/configuration.go b/internal/controller/admin/configuration.go new file mode 100644 index 0000000..2cd7cf0 --- /dev/null +++ b/internal/controller/admin/configuration.go @@ -0,0 +1,84 @@ +package admin + +import ( + "bytes" + "context" + "errors" + "fmt" + "reflect" + "strings" + + "proxy-pool/internal/config" + "proxy-pool/internal/domain/adminstate" +) + +var ErrInvalidConfigurationLoader = errors.New("invalid admin configuration loader") + +type FileConfigurationLoader struct { + path string + resolver config.Resolver +} + +func NewFileConfigurationLoader(path string, resolver config.Resolver) (*FileConfigurationLoader, error) { + if strings.TrimSpace(path) != path || path == "" || len(path) > adminstate.MaxSourceBytes || nilInterface(resolver) { + return nil, ErrInvalidConfigurationLoader + } + return &FileConfigurationLoader{path: path, resolver: resolver}, nil +} + +func (loader *FileConfigurationLoader) LoadConfiguration(ctx context.Context) (LoadedConfiguration, error) { + if loader == nil || ctx == nil { + return LoadedConfiguration{}, ErrInvalidConfigurationLoader + } + if err := ctx.Err(); err != nil { + return LoadedConfiguration{}, err + } + content, err := loader.resolver.ReadFile(loader.path) + if err != nil { + return LoadedConfiguration{}, errors.Join(ErrUnavailable, + fmt.Errorf("read configuration %q: %w", loader.path, err)) + } + if err := ctx.Err(); err != nil { + return LoadedConfiguration{}, err + } + resolver := &observedConfigurationResolver{Resolver: loader.resolver} + configuration, err := config.LoadResolved(bytes.NewReader(content), resolver) + if err != nil { + if resolver.readErr != nil { + return LoadedConfiguration{}, errors.Join(ErrUnavailable, + fmt.Errorf("load configuration %q: %w", loader.path, err)) + } + return LoadedConfiguration{}, errors.Join(ErrInvalidConfiguration, + fmt.Errorf("load configuration %q: %w", loader.path, err)) + } + if err := ctx.Err(); err != nil { + return LoadedConfiguration{}, err + } + return LoadedConfiguration{Value: configuration, Source: loader.path}, nil +} + +type observedConfigurationResolver struct { + config.Resolver + readErr error +} + +func (resolver *observedConfigurationResolver) ReadFile(path string) ([]byte, error) { + content, err := resolver.Resolver.ReadFile(path) + if err != nil { + resolver.readErr = err + } + return content, err +} + +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/admin/configuration_test.go b/internal/controller/admin/configuration_test.go new file mode 100644 index 0000000..e104e95 --- /dev/null +++ b/internal/controller/admin/configuration_test.go @@ -0,0 +1,156 @@ +package admin + +import ( + "context" + "errors" + "strings" + "testing" +) + +func TestFileConfigurationLoaderUsesStrictResolvedLoader(t *testing.T) { + t.Parallel() + configuration := strings.Replace(fileLoaderConfiguration, "proxyAuth: {type: response}", + "proxyAuth: {type: static, username: alice, passwordFile: /run/secrets/proxy}", 1) + resolver := &memoryConfigurationResolver{files: map[string][]byte{ + "configs/proxy-pool.yaml": []byte(configuration), + "/run/secrets/proxy": []byte("resolved-secret\r\n"), + }} + loader, err := NewFileConfigurationLoader("configs/proxy-pool.yaml", resolver) + if err != nil { + t.Fatalf("NewFileConfigurationLoader() error = %v", err) + } + + loaded, err := loader.LoadConfiguration(context.Background()) + if err != nil { + t.Fatalf("LoadConfiguration() error = %v", err) + } + if loaded.Source != "configs/proxy-pool.yaml" || loaded.Value == nil { + t.Fatalf("LoadConfiguration() = %+v", loaded) + } + auth := loaded.Value.Upstreams["provider-a"].ProxyAuth + if auth.Password != "resolved-secret" || auth.PasswordFile != "" { + t.Fatalf("resolved proxy auth = %+v", auth) + } +} + +func TestFileConfigurationLoaderRejectsUnknownFields(t *testing.T) { + t.Parallel() + resolver := &memoryConfigurationResolver{files: map[string][]byte{ + "config.yaml": []byte(fileLoaderConfiguration + "\nunknownRootField: true\n"), + }} + loader, err := NewFileConfigurationLoader("config.yaml", resolver) + if err != nil { + t.Fatalf("NewFileConfigurationLoader() error = %v", err) + } + if _, err := loader.LoadConfiguration(context.Background()); !errors.Is(err, ErrInvalidConfiguration) || !strings.Contains(err.Error(), "unknownRootField") { + t.Fatalf("LoadConfiguration() error = %v, want strict unknown field error", err) + } +} + +func TestFileConfigurationLoaderClassifiesReadFailureAsUnavailable(t *testing.T) { + t.Parallel() + loader, err := NewFileConfigurationLoader("missing.yaml", &memoryConfigurationResolver{}) + if err != nil { + t.Fatalf("NewFileConfigurationLoader() error = %v", err) + } + if _, err := loader.LoadConfiguration(context.Background()); !errors.Is(err, ErrUnavailable) { + t.Fatalf("LoadConfiguration() error = %v, want %v", err, ErrUnavailable) + } +} + +func TestFileConfigurationLoaderClassifiesSecretReadFailureAsUnavailable(t *testing.T) { + t.Parallel() + configuration := strings.Replace(fileLoaderConfiguration, "proxyAuth: {type: response}", + "proxyAuth: {type: static, username: alice, passwordFile: /run/secrets/missing}", 1) + resolver := &memoryConfigurationResolver{files: map[string][]byte{ + "config.yaml": []byte(configuration), + }} + loader, err := NewFileConfigurationLoader("config.yaml", resolver) + if err != nil { + t.Fatalf("NewFileConfigurationLoader() error = %v", err) + } + if _, err := loader.LoadConfiguration(context.Background()); !errors.Is(err, ErrUnavailable) || errors.Is(err, ErrInvalidConfiguration) { + t.Fatalf("LoadConfiguration() error = %v, want unavailable secret source", err) + } +} + +func TestFileConfigurationLoaderHonorsCancellationBeforeIO(t *testing.T) { + t.Parallel() + resolver := &memoryConfigurationResolver{} + loader, err := NewFileConfigurationLoader("config.yaml", resolver) + if err != nil { + t.Fatalf("NewFileConfigurationLoader() error = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := loader.LoadConfiguration(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("LoadConfiguration() error = %v, want context cancellation", err) + } + if resolver.reads != 0 { + t.Fatalf("resolver reads = %d, want 0", resolver.reads) + } +} + +func TestNewFileConfigurationLoaderRejectsInvalidDependencies(t *testing.T) { + t.Parallel() + resolver := &memoryConfigurationResolver{} + for _, test := range []struct { + path string + resolver *memoryConfigurationResolver + }{ + {resolver: resolver}, + {path: "config.yaml"}, + } { + if _, err := NewFileConfigurationLoader(test.path, test.resolver); !errors.Is(err, ErrInvalidConfigurationLoader) { + t.Fatalf("NewFileConfigurationLoader(%q) error = %v", test.path, err) + } + } +} + +type memoryConfigurationResolver struct { + files map[string][]byte + reads int +} + +func (*memoryConfigurationResolver) LookupEnv(string) (string, bool) { return "", false } + +func (resolver *memoryConfigurationResolver) ReadFile(path string) ([]byte, error) { + resolver.reads++ + content, exists := resolver.files[path] + if !exists { + return nil, errors.New("fixture file not found") + } + return append([]byte(nil), content...), nil +} + +const fileLoaderConfiguration = ` +version: 1 +security: {requireProtectionOnPublicListen: true} +gateway: + enabled: true + listen: 127.0.0.1:8080 + auth: {mode: none} + limits: {maxConcurrentConnections: 20000} + retry: {maxAttempts: 2, retryMethods: [GET, HEAD]} + destinationPolicy: {denyPrivateNetworks: true, denyLoopback: true, denyLinkLocal: true} +routing: + - name: gateway + enabled: true + purpose: gateway + match: {hostRegex: '.*'} + upstreams: [provider-a] + strategy: {type: leastConnections} + onUnavailable: {action: reject} +upstreams: + provider-a: + enabled: true + exposure: [gateway] + provider: {billingMode: subscription, protocols: [http, https]} + api: {url: https://provider-a.example/proxies, method: GET, auth: {type: none}, template: '{{.}}'} + proxyAuth: {type: response} + pool: {maxSize: 2000} + capacity: {maxConcurrencyPerProxy: 20} + lifecycle: {ttl: 5m, allocationSafetyMargin: 30s} + fetch: {requestInterval: 1s, timeout: 3s, maxAttempts: 3, maxInFlight: 1} + check: {interval: 30s, jitter: 20, maxInFlight: 100, timeout: 2s, maxAttempts: 2, maxConsecutiveFailures: 3} +` diff --git a/internal/controller/admin/service.go b/internal/controller/admin/service.go new file mode 100644 index 0000000..fc9e2db --- /dev/null +++ b/internal/controller/admin/service.go @@ -0,0 +1,281 @@ +package admin + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "sort" + "strings" + "time" + + "proxy-pool/internal/config" + "proxy-pool/internal/domain/adminstate" +) + +var ErrInvalidApplicationService = errors.New("invalid admin application service") + +type StateRepository interface { + adminstate.Mutator + adminstate.SnapshotReader +} + +type OperationalStatusReader interface { + ReadOperationalStatus(context.Context) (OperationalStatus, error) +} + +type ConfigurationLoader interface { + LoadConfiguration(context.Context) (LoadedConfiguration, error) +} + +// ConfigurationPublisher must atomically publish an already validated configuration. +type ConfigurationPublisher interface { + Publish(*config.Config) +} + +var _ ConfigurationPublisher = (*config.Store)(nil) + +type ApplicationDependencies struct { + State StateRepository + Operations OperationalStatusReader + Configuration ConfigurationLoader + Publisher ConfigurationPublisher +} + +type ApplicationOptions struct { + Now func() time.Time +} + +type OperationalStatus struct { + SnapshotVersion uint64 + Upstreams []UpstreamActivity + Workers []WorkerStatus +} + +type UpstreamActivity struct { + Name string + Available int64 + Checking int64 + Suspect int64 + Draining int64 + Extracted int64 + ConsecutiveEmptyFetch int64 + FetchErrorCount int64 +} + +type LoadedConfiguration struct { + Value *config.Config + Source string +} + +type ApplicationService struct { + state StateRepository + operations OperationalStatusReader + configuration ConfigurationLoader + publisher ConfigurationPublisher + now func() time.Time +} + +var _ Service = (*ApplicationService)(nil) + +func NewApplicationService(dependencies ApplicationDependencies, options ApplicationOptions) (*ApplicationService, error) { + if nilInterface(dependencies.State) || nilInterface(dependencies.Operations) || nilInterface(dependencies.Configuration) || + nilInterface(dependencies.Publisher) || options.Now == nil { + return nil, ErrInvalidApplicationService + } + return &ApplicationService{ + state: dependencies.State, + operations: dependencies.Operations, + configuration: dependencies.Configuration, + publisher: dependencies.Publisher, + now: options.Now, + }, nil +} + +func (service *ApplicationService) SetUpstreamEnabled(ctx context.Context, command SetUpstreamCommand) (MutationResult, error) { + result, err := service.state.SetUpstreamEnabled(ctx, adminstate.SetUpstreamCommand{ + RequestID: command.RequestID, + Actor: adminstate.Actor{ID: command.ActorID, SourceIP: command.SourceIP}, + OccurredAt: service.now().UTC(), + Name: command.Name, + Enabled: command.Enabled, + }) + return mutationResult(result), mapAdminStateError(err) +} + +func (service *ApplicationService) SwitchRouting(ctx context.Context, command SwitchCommand) (MutationResult, error) { + result, err := service.state.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{ + RequestID: command.RequestID, + Actor: adminstate.Actor{ID: command.ActorID, SourceIP: command.SourceIP}, + OccurredAt: service.now().UTC(), + Name: command.Name, + ExpectedCurrent: command.ExpectedCurrent, + Target: command.Target, + Reason: command.Reason, + }) + return mutationResult(result), mapAdminStateError(err) +} + +func (service *ApplicationService) Status(ctx context.Context) (Status, error) { + snapshot, err := service.state.Snapshot(ctx) + if err != nil { + return Status{}, mapAdminStateError(err) + } + operations, err := service.operations.ReadOperationalStatus(ctx) + if err != nil { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return Status{}, err + } + return Status{}, errors.Join(ErrUnavailable, err) + } + + activityByName := make(map[string]UpstreamActivity, len(operations.Upstreams)) + for _, activity := range operations.Upstreams { + activityByName[activity.Name] = activity + } + status := Status{SnapshotVersion: operations.SnapshotVersion} + if snapshot.Config != nil { + status.ConfigVersion = snapshot.Config.ConfigVersion + } + status.Upstreams = make([]UpstreamStatus, 0, len(snapshot.Upstreams)) + for _, authoritative := range snapshot.Upstreams { + activity := activityByName[authoritative.Name] + status.Upstreams = append(status.Upstreams, UpstreamStatus{ + Name: authoritative.Name, + Enabled: authoritative.Enabled, + Available: activity.Available, + Checking: activity.Checking, + Suspect: activity.Suspect, + Draining: activity.Draining, + Extracted: activity.Extracted, + ConsecutiveEmptyFetch: activity.ConsecutiveEmptyFetch, + FetchErrorCount: activity.FetchErrorCount, + }) + } + sort.Slice(status.Upstreams, func(left, right int) bool { + return status.Upstreams[left].Name < status.Upstreams[right].Name + }) + status.Workers = append([]WorkerStatus(nil), operations.Workers...) + sort.Slice(status.Workers, func(left, right int) bool { + if status.Workers[left].ID == status.Workers[right].ID { + return status.Workers[left].Zone < status.Workers[right].Zone + } + return status.Workers[left].ID < status.Workers[right].ID + }) + return status, nil +} + +func (service *ApplicationService) ReloadConfiguration(ctx context.Context, command ReloadCommand) (MutationResult, error) { + loaded, err := service.configuration.LoadConfiguration(ctx) + if err != nil { + switch { + case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded): + return MutationResult{RequestID: command.RequestID}, err + case errors.Is(err, ErrUnavailable), errors.Is(err, ErrInvalidConfiguration): + return MutationResult{RequestID: command.RequestID}, err + default: + return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err) + } + } + if loaded.Value == nil || strings.TrimSpace(loaded.Source) != loaded.Source || loaded.Source == "" || + len(loaded.Source) > adminstate.MaxSourceBytes { + return MutationResult{RequestID: command.RequestID}, ErrInvalidConfiguration + } + if err := config.Validate(loaded.Value); err != nil { + return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err) + } + + managementView := loaded.Value.Redacted() + encoded, err := json.Marshal(managementView) + if err != nil { + return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err) + } + digest := sha256.Sum256(encoded) + checksum := hex.EncodeToString(digest[:]) + + current, err := service.state.Snapshot(ctx) + if err != nil { + return MutationResult{RequestID: command.RequestID}, mapAdminStateError(err) + } + upstreams, routings := managementDefinitions(loaded.Value, current) + result, err := service.state.CommitConfig(ctx, adminstate.CommitConfigCommand{ + RequestID: command.RequestID, + Actor: adminstate.Actor{ID: command.ActorID, SourceIP: command.SourceIP}, + OccurredAt: service.now().UTC(), + ConfigVersion: "cfg-" + checksum, + Checksum: checksum, + Source: loaded.Source, + Upstreams: upstreams, + Routings: routings, + }) + if err != nil { + return mutationResult(result), mapAdminStateError(err) + } + service.publisher.Publish(loaded.Value) + return mutationResult(result), nil +} + +func managementDefinitions(configuration *config.Config, current adminstate.Snapshot) ([]adminstate.UpstreamDefinition, []adminstate.RoutingDefinition) { + upstreams := make([]adminstate.UpstreamDefinition, 0, len(configuration.Upstreams)) + for name, upstream := range configuration.Upstreams { + upstreams = append(upstreams, adminstate.UpstreamDefinition{Name: name, Enabled: upstream.Enabled}) + } + sort.Slice(upstreams, func(left, right int) bool { return upstreams[left].Name < upstreams[right].Name }) + + currentByName := make(map[string]string, len(current.Routings)) + for _, routing := range current.Routings { + currentByName[routing.Name] = routing.CurrentUpstream + } + routings := make([]adminstate.RoutingDefinition, 0, len(configuration.Routing)) + for _, routing := range configuration.Routing { + selected := "" + if existing := currentByName[routing.Name]; containsString(routing.Upstreams, existing) { + selected = existing + } else if len(routing.Upstreams) > 0 { + selected = routing.Upstreams[0] + } + routings = append(routings, adminstate.RoutingDefinition{ + Name: routing.Name, Enabled: routing.Enabled, + Upstreams: append([]string(nil), routing.Upstreams...), CurrentUpstream: selected, + }) + } + sort.Slice(routings, func(left, right int) bool { return routings[left].Name < routings[right].Name }) + return upstreams, routings +} + +func containsString(values []string, target string) bool { + for _, value := range values { + if value == target { + return true + } + } + return false +} + +func mutationResult(result adminstate.MutationResult) MutationResult { + return MutationResult{ + RequestID: result.RequestID, + Changed: result.Changed, + Version: result.Revision, + Message: result.Message, + } +} + +func mapAdminStateError(err error) error { + if err == nil { + return nil + } + switch { + case errors.Is(err, adminstate.ErrNotFound): + return errors.Join(ErrNotFound, err) + case errors.Is(err, adminstate.ErrConflict): + return errors.Join(ErrConflict, err) + case errors.Is(err, adminstate.ErrInvalidCommand): + return errors.Join(ErrInvalidConfiguration, err) + case errors.Is(err, adminstate.ErrUnavailable): + return errors.Join(ErrUnavailable, err) + default: + return err + } +} diff --git a/internal/controller/admin/service_boundary_test.go b/internal/controller/admin/service_boundary_test.go new file mode 100644 index 0000000..520d503 --- /dev/null +++ b/internal/controller/admin/service_boundary_test.go @@ -0,0 +1,43 @@ +package admin + +import ( + "go/ast" + "go/parser" + "go/token" + "strconv" + "strings" + "testing" +) + +func TestAdminApplicationHasNoProxyDetailOrRedisExtractionDependency(t *testing.T) { + t.Parallel() + packages, err := parser.ParseDir(token.NewFileSet(), ".", nil, parser.ImportsOnly) + if err != nil { + t.Fatalf("ParseDir(): %v", err) + } + for _, parsedPackage := range packages { + for filename, file := range parsedPackage.Files { + ast.Inspect(file, func(node ast.Node) bool { + importSpec, ok := node.(*ast.ImportSpec) + if !ok { + return true + } + path, err := strconv.Unquote(importSpec.Path.Value) + if err != nil { + t.Errorf("unquote import in %s: %v", filename, err) + return false + } + for _, forbidden := range []string{ + "internal/adapters/redisactivity", + "internal/domain/activitypool", + "internal/domain/extraction", + } { + if strings.Contains(path, forbidden) { + t.Errorf("%s imports forbidden data-plane package %q", filename, path) + } + } + return false + }) + } + } +} diff --git a/internal/controller/admin/service_test.go b/internal/controller/admin/service_test.go new file mode 100644 index 0000000..ee4c092 --- /dev/null +++ b/internal/controller/admin/service_test.go @@ -0,0 +1,513 @@ +package admin + +import ( + "context" + "errors" + "testing" + "time" + + "proxy-pool/internal/config" + "proxy-pool/internal/domain/adminstate" +) + +func TestApplicationServiceMapsUpstreamMutationToAdminState(t *testing.T) { + t.Parallel() + now := time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC) + state := &recordingAdminState{ + mutation: adminstate.MutationResult{ + RequestID: "req-enable", + Changed: true, + Revision: 12, + Message: "enabled", + }, + } + service, err := NewApplicationService(ApplicationDependencies{ + State: state, + Operations: staticOperationalStatusReader{}, + Configuration: staticConfigurationLoader{}, + Publisher: &recordingConfigurationPublisher{}, + }, ApplicationOptions{Now: func() time.Time { return now }}) + if err != nil { + t.Fatalf("NewApplicationService() error = %v", err) + } + + result, err := service.SetUpstreamEnabled(context.Background(), SetUpstreamCommand{ + RequestID: "req-enable", + ActorID: "admin:alice", + SourceIP: "192.0.2.10", + Name: "provider-a", + Enabled: true, + }) + if err != nil { + t.Fatalf("SetUpstreamEnabled() error = %v", err) + } + if result != (MutationResult{RequestID: "req-enable", Changed: true, Version: 12, Message: "enabled"}) { + t.Fatalf("SetUpstreamEnabled() result = %+v", result) + } + if state.lastUpstream != (adminstate.SetUpstreamCommand{ + RequestID: "req-enable", + Actor: adminstate.Actor{ID: "admin:alice", SourceIP: "192.0.2.10"}, + OccurredAt: now, + Name: "provider-a", + Enabled: true, + }) { + t.Fatalf("admin state command = %+v", state.lastUpstream) + } +} + +func TestNewApplicationServiceRejectsMissingDependencies(t *testing.T) { + t.Parallel() + valid := ApplicationDependencies{ + State: &recordingAdminState{}, + Operations: staticOperationalStatusReader{}, + Configuration: staticConfigurationLoader{}, + Publisher: &recordingConfigurationPublisher{}, + } + tests := []struct { + name string + dependencies ApplicationDependencies + options ApplicationOptions + }{ + {name: "state", dependencies: func() ApplicationDependencies { value := valid; value.State = nil; return value }(), options: ApplicationOptions{Now: time.Now}}, + {name: "operations", dependencies: func() ApplicationDependencies { value := valid; value.Operations = nil; return value }(), options: ApplicationOptions{Now: time.Now}}, + {name: "configuration", dependencies: func() ApplicationDependencies { value := valid; value.Configuration = nil; return value }(), options: ApplicationOptions{Now: time.Now}}, + {name: "publisher", dependencies: func() ApplicationDependencies { value := valid; value.Publisher = nil; return value }(), options: ApplicationOptions{Now: time.Now}}, + {name: "clock", dependencies: valid}, + } + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + if _, err := NewApplicationService(test.dependencies, test.options); !errors.Is(err, ErrInvalidApplicationService) { + t.Fatalf("NewApplicationService() error = %v, want %v", err, ErrInvalidApplicationService) + } + }) + } +} + +func TestNewApplicationServiceRejectsTypedNilDependencies(t *testing.T) { + t.Parallel() + var state *recordingAdminState + var publisher *recordingConfigurationPublisher + valid := ApplicationDependencies{ + State: &recordingAdminState{}, + Operations: staticOperationalStatusReader{}, + Configuration: staticConfigurationLoader{}, + Publisher: &recordingConfigurationPublisher{}, + } + for _, dependencies := range []ApplicationDependencies{ + func() ApplicationDependencies { value := valid; value.State = state; return value }(), + func() ApplicationDependencies { value := valid; value.Publisher = publisher; return value }(), + } { + if _, err := NewApplicationService(dependencies, ApplicationOptions{Now: time.Now}); !errors.Is(err, ErrInvalidApplicationService) { + t.Fatalf("NewApplicationService(typed nil) error = %v, want %v", err, ErrInvalidApplicationService) + } + } +} + +func TestApplicationServiceMapsRoutingSwitchAndDomainErrors(t *testing.T) { + t.Parallel() + now := time.Date(2026, 7, 29, 11, 0, 0, 0, time.FixedZone("test", 8*60*60)) + state := &recordingAdminState{mutation: adminstate.MutationResult{RequestID: "req-switch", Changed: true, Revision: 21}} + service := mustApplicationService(t, state, ApplicationOptions{Now: func() time.Time { return now }}) + + result, err := service.SwitchRouting(context.Background(), SwitchCommand{ + RequestID: "req-switch", ActorID: "admin:bob", SourceIP: "198.51.100.7", + Name: "checkout", ExpectedCurrent: "provider-a", Target: "provider-b", Reason: "capacity", + }) + if err != nil { + t.Fatalf("SwitchRouting() error = %v", err) + } + if result.Version != 21 || !result.Changed { + t.Fatalf("SwitchRouting() result = %+v", result) + } + wantCommand := adminstate.SwitchRoutingCommand{ + RequestID: "req-switch", Actor: adminstate.Actor{ID: "admin:bob", SourceIP: "198.51.100.7"}, + OccurredAt: now.UTC(), Name: "checkout", ExpectedCurrent: "provider-a", Target: "provider-b", Reason: "capacity", + } + if state.lastSwitch != wantCommand { + t.Fatalf("admin state command = %+v, want %+v", state.lastSwitch, wantCommand) + } + + tests := []struct { + domain error + want error + }{ + {domain: adminstate.ErrNotFound, want: ErrNotFound}, + {domain: adminstate.ErrConflict, want: ErrConflict}, + {domain: adminstate.ErrInvalidCommand, want: ErrInvalidConfiguration}, + {domain: adminstate.ErrUnavailable, want: ErrUnavailable}, + } + for _, test := range tests { + state.err = test.domain + _, err := service.SwitchRouting(context.Background(), SwitchCommand{ + RequestID: "req-switch", ActorID: "admin:bob", SourceIP: "198.51.100.7", + Name: "checkout", ExpectedCurrent: "provider-a", Target: "provider-b", + }) + if !errors.Is(err, test.want) || !errors.Is(err, test.domain) { + t.Fatalf("SwitchRouting(%v) error = %v, want mapped %v preserving cause", test.domain, err, test.want) + } + } +} + +func TestApplicationServiceBuildsStatusFromAuthoritativeAndOperationalSnapshots(t *testing.T) { + t.Parallel() + state := &recordingAdminState{snapshot: adminstate.Snapshot{ + Revision: 31, + Config: &adminstate.ConfigRevision{Revision: 31, ConfigVersion: "cfg-31"}, + Upstreams: []adminstate.UpstreamState{ + {Name: "provider-b", Enabled: false, Revision: 31}, + {Name: "provider-a", Enabled: true, Revision: 31}, + }, + }} + operations := staticOperationalStatusReader{status: OperationalStatus{ + SnapshotVersion: 88, + Upstreams: []UpstreamActivity{ + {Name: "provider-a", Available: 10, Checking: 2, Suspect: 1, ConsecutiveEmptyFetch: 3}, + {Name: "unknown", Available: 999}, + }, + Workers: []WorkerStatus{ + {ID: "worker-b", Zone: "zone-b", Connected: false, SnapshotVersion: 87, StaleSeconds: 4}, + {ID: "worker-a", Zone: "zone-a", Connected: true, SnapshotVersion: 88}, + }, + }} + service, err := NewApplicationService(ApplicationDependencies{ + State: state, Operations: operations, Configuration: staticConfigurationLoader{}, + Publisher: &recordingConfigurationPublisher{}, + }, ApplicationOptions{Now: time.Now}) + if err != nil { + t.Fatalf("NewApplicationService() error = %v", err) + } + + status, err := service.Status(context.Background()) + if err != nil { + t.Fatalf("Status() error = %v", err) + } + if status.ConfigVersion != "cfg-31" || status.SnapshotVersion != 88 { + t.Fatalf("Status() versions = (%q, %d)", status.ConfigVersion, status.SnapshotVersion) + } + if len(status.Upstreams) != 2 || status.Upstreams[0].Name != "provider-a" || !status.Upstreams[0].Enabled || + status.Upstreams[0].Available != 10 || status.Upstreams[0].Checking != 2 || + status.Upstreams[0].ConsecutiveEmptyFetch != 3 { + t.Fatalf("Status() upstreams = %+v", status.Upstreams) + } + if status.Upstreams[1].Name != "provider-b" || status.Upstreams[1].Enabled || status.Upstreams[1].Available != 0 { + t.Fatalf("Status() disabled upstream = %+v", status.Upstreams[1]) + } + if len(status.Workers) != 2 || status.Workers[0].ID != "worker-a" || status.Workers[1].ID != "worker-b" { + t.Fatalf("Status() workers = %+v", status.Workers) + } +} + +func TestApplicationServiceMapsStatusDependencyFailures(t *testing.T) { + t.Parallel() + stateFailure := errors.Join(adminstate.ErrUnavailable, errors.New("postgres down")) + service, err := NewApplicationService(ApplicationDependencies{ + State: &recordingAdminState{err: stateFailure}, Operations: staticOperationalStatusReader{}, + Configuration: staticConfigurationLoader{}, Publisher: &recordingConfigurationPublisher{}, + }, ApplicationOptions{Now: time.Now}) + if err != nil { + t.Fatalf("NewApplicationService() error = %v", err) + } + if _, err := service.Status(context.Background()); !errors.Is(err, ErrUnavailable) || !errors.Is(err, stateFailure) { + t.Fatalf("Status(state failure) error = %v", err) + } + + operationsFailure := errors.New("redis aggregate unavailable") + service, err = NewApplicationService(ApplicationDependencies{ + State: &recordingAdminState{}, Operations: staticOperationalStatusReader{err: operationsFailure}, + Configuration: staticConfigurationLoader{}, Publisher: &recordingConfigurationPublisher{}, + }, ApplicationOptions{Now: time.Now}) + if err != nil { + t.Fatalf("NewApplicationService() error = %v", err) + } + if _, err := service.Status(context.Background()); !errors.Is(err, ErrUnavailable) || !errors.Is(err, operationsFailure) { + t.Fatalf("Status(operations failure) error = %v", err) + } + + service, err = NewApplicationService(ApplicationDependencies{ + State: &recordingAdminState{}, Operations: staticOperationalStatusReader{err: context.Canceled}, + Configuration: staticConfigurationLoader{}, Publisher: &recordingConfigurationPublisher{}, + }, ApplicationOptions{Now: time.Now}) + if err != nil { + t.Fatalf("NewApplicationService() error = %v", err) + } + if _, err := service.Status(context.Background()); !errors.Is(err, context.Canceled) || errors.Is(err, ErrUnavailable) { + t.Fatalf("Status(cancellation) error = %v, want unclassified cancellation", err) + } +} + +func TestApplicationServiceReloadPersistsManagementViewBeforePublishing(t *testing.T) { + t.Parallel() + now := time.Date(2026, 7, 29, 12, 0, 0, 0, time.UTC) + configuration := validReloadConfiguration() + publisher := &recordingConfigurationPublisher{} + state := &recordingAdminState{ + mutation: adminstate.MutationResult{RequestID: "req-reload", Changed: true, Revision: 42}, + snapshot: adminstate.Snapshot{Routings: []adminstate.RoutingState{ + {Name: "checkout", Enabled: true, Upstreams: []string{"provider-a", "provider-b"}, CurrentUpstream: "provider-b"}, + }}, + } + state.onCommit = func(adminstate.CommitConfigCommand) { + if len(publisher.published) != 0 { + t.Fatal("configuration was published before management state committed") + } + } + service, err := NewApplicationService(ApplicationDependencies{ + State: state, Operations: staticOperationalStatusReader{}, + Configuration: staticConfigurationLoader{loaded: LoadedConfiguration{ + Value: configuration, Source: "configs/proxy-pool.yaml", + }}, + Publisher: publisher, + }, ApplicationOptions{Now: func() time.Time { return now }}) + if err != nil { + t.Fatalf("NewApplicationService() error = %v", err) + } + + result, err := service.ReloadConfiguration(context.Background(), ReloadCommand{ + RequestID: "req-reload", ActorID: "admin:alice", SourceIP: "192.0.2.10", + }) + if err != nil { + t.Fatalf("ReloadConfiguration() error = %v", err) + } + if result.Version != 42 || !result.Changed { + t.Fatalf("ReloadConfiguration() result = %+v", result) + } + if len(publisher.published) != 1 || publisher.published[0] != configuration { + t.Fatalf("published configurations = %+v", publisher.published) + } + command := state.lastConfig + if command.RequestID != "req-reload" || command.Actor != (adminstate.Actor{ID: "admin:alice", SourceIP: "192.0.2.10"}) || + !command.OccurredAt.Equal(now) || command.Source != "configs/proxy-pool.yaml" { + t.Fatalf("CommitConfig() metadata = %+v", command) + } + if len(command.Checksum) != adminstate.SHA256HexBytes || command.ConfigVersion != "cfg-"+command.Checksum { + t.Fatalf("CommitConfig() version/checksum = (%q, %q)", command.ConfigVersion, command.Checksum) + } + if len(command.Upstreams) != 2 || command.Upstreams[0].Name != "provider-a" || command.Upstreams[1].Name != "provider-b" { + t.Fatalf("CommitConfig() upstreams = %+v", command.Upstreams) + } + if len(command.Routings) != 2 || command.Routings[0].Name != "checkout" || command.Routings[0].CurrentUpstream != "provider-b" || + command.Routings[1].Name != "new-route" || command.Routings[1].CurrentUpstream != "provider-b" { + t.Fatalf("CommitConfig() routings = %+v", command.Routings) + } +} + +func TestApplicationServiceReloadDoesNotPublishInvalidOrUncommittedConfiguration(t *testing.T) { + t.Parallel() + tests := []struct { + name string + loaded LoadedConfiguration + stateError error + wantCause error + }{ + { + name: "invalid configuration", loaded: LoadedConfiguration{Value: &config.Config{}, Source: "invalid.yaml"}, + wantCause: ErrInvalidConfiguration, + }, + { + name: "persistence unavailable", loaded: LoadedConfiguration{Value: validReloadConfiguration(), Source: "valid.yaml"}, + stateError: adminstate.ErrUnavailable, wantCause: ErrUnavailable, + }, + } + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + publisher := &recordingConfigurationPublisher{} + state := &recordingAdminState{err: test.stateError} + service, err := NewApplicationService(ApplicationDependencies{ + State: state, Operations: staticOperationalStatusReader{}, + Configuration: staticConfigurationLoader{loaded: test.loaded}, Publisher: publisher, + }, ApplicationOptions{Now: time.Now}) + if err != nil { + t.Fatalf("NewApplicationService() error = %v", err) + } + _, err = service.ReloadConfiguration(context.Background(), ReloadCommand{ + RequestID: "req-reload", ActorID: "admin:alice", SourceIP: "192.0.2.10", + }) + if !errors.Is(err, test.wantCause) { + t.Fatalf("ReloadConfiguration() error = %v, want %v", err, test.wantCause) + } + if len(publisher.published) != 0 { + t.Fatalf("published %d configurations after failure", len(publisher.published)) + } + }) + } +} + +func TestApplicationServiceReloadPublishesSuccessfulReplay(t *testing.T) { + t.Parallel() + publisher := &recordingConfigurationPublisher{} + state := &recordingAdminState{mutation: adminstate.MutationResult{RequestID: "req-replay", Revision: 7}} + service, err := NewApplicationService(ApplicationDependencies{ + State: state, Operations: staticOperationalStatusReader{}, + Configuration: staticConfigurationLoader{loaded: LoadedConfiguration{ + Value: validReloadConfiguration(), Source: "config.yaml", + }}, Publisher: publisher, + }, ApplicationOptions{Now: time.Now}) + if err != nil { + t.Fatalf("NewApplicationService() error = %v", err) + } + result, err := service.ReloadConfiguration(context.Background(), ReloadCommand{ + RequestID: "req-replay", ActorID: "admin:alice", SourceIP: "192.0.2.10", + }) + if err != nil || result.Changed || result.Version != 7 || len(publisher.published) != 1 { + t.Fatalf("ReloadConfiguration() = %+v, %v; publishes=%d", result, err, len(publisher.published)) + } +} + +func TestApplicationServiceReloadPreservesCancellation(t *testing.T) { + t.Parallel() + service, err := NewApplicationService(ApplicationDependencies{ + State: &recordingAdminState{}, Operations: staticOperationalStatusReader{}, + Configuration: staticConfigurationLoader{err: context.Canceled}, + Publisher: &recordingConfigurationPublisher{}, + }, ApplicationOptions{Now: time.Now}) + if err != nil { + t.Fatalf("NewApplicationService() error = %v", err) + } + _, err = service.ReloadConfiguration(context.Background(), ReloadCommand{RequestID: "req-cancel"}) + if !errors.Is(err, context.Canceled) || errors.Is(err, ErrInvalidConfiguration) { + t.Fatalf("ReloadConfiguration() error = %v, want unclassified cancellation", err) + } +} + +func TestApplicationServiceUsesSecretFreeManagementChecksum(t *testing.T) { + t.Parallel() + state := &recordingAdminState{} + publisher := &recordingConfigurationPublisher{} + var commands []adminstate.CommitConfigCommand + state.onCommit = func(command adminstate.CommitConfigCommand) { + commands = append(commands, command) + } + for _, secret := range []string{"secret-a", "secret-b"} { + configuration := validReloadConfiguration() + upstream := configuration.Upstreams["provider-a"] + upstream.ProxyAuth.Password = secret + configuration.Upstreams["provider-a"] = upstream + service, err := NewApplicationService(ApplicationDependencies{ + State: state, Operations: staticOperationalStatusReader{}, + Configuration: staticConfigurationLoader{loaded: LoadedConfiguration{ + Value: configuration, Source: "config.yaml", + }}, Publisher: publisher, + }, ApplicationOptions{Now: time.Now}) + if err != nil { + t.Fatalf("NewApplicationService() error = %v", err) + } + if _, err := service.ReloadConfiguration(context.Background(), ReloadCommand{ + RequestID: "req-secret", ActorID: "admin:alice", SourceIP: "192.0.2.10", + }); err != nil { + t.Fatalf("ReloadConfiguration() error = %v", err) + } + } + if len(commands) != 2 || commands[0].Checksum != commands[1].Checksum || commands[0].ConfigVersion != commands[1].ConfigVersion { + t.Fatalf("secret rotation changed public management digest: %+v", commands) + } + if len(publisher.published) != 2 || publisher.published[1].Upstreams["provider-a"].ProxyAuth.Password != "secret-b" { + t.Fatalf("secret rotation was not published: %+v", publisher.published) + } +} + +func validReloadConfiguration() *config.Config { + return &config.Config{ + Version: 1, + Upstreams: map[string]config.Upstream{ + "provider-b": validReloadUpstream("secret-b"), + "provider-a": validReloadUpstream("secret-a"), + }, + Routing: []config.Routing{ + { + Name: "new-route", Enabled: true, Purpose: "gateway", + Upstreams: []string{"provider-b", "provider-a"}, Strategy: config.Strategy{Type: "random"}, + OnUnavailable: config.OnUnavailable{Action: "reject"}, + }, + { + Name: "checkout", Enabled: true, Purpose: "gateway", + Upstreams: []string{"provider-a", "provider-b"}, Strategy: config.Strategy{Type: "sequential", SwitchAfterEmptyFetch: 5}, + OnUnavailable: config.OnUnavailable{Action: "reject"}, + }, + }, + } +} + +func validReloadUpstream(secret string) config.Upstream { + return config.Upstream{ + Enabled: true, Exposure: []string{"gateway"}, + API: config.ProviderAPI{Auth: config.ProviderAuth{Type: "none"}}, + ProxyAuth: config.ProxyAuth{Type: "static", Username: "user", Password: secret}, + Pool: config.Pool{MaxSize: 10}, Capacity: config.Capacity{MaxConcurrencyPerProxy: 2}, + Lifecycle: config.Lifecycle{TTL: config.Duration(time.Minute), AllocationSafetyMargin: config.Duration(10 * time.Second)}, + Fetch: config.Fetch{Timeout: config.Duration(time.Second), MaxAttempts: 2, MaxInFlight: 1}, + } +} + +func mustApplicationService(t *testing.T, state StateRepository, options ApplicationOptions) *ApplicationService { + t.Helper() + service, err := NewApplicationService(ApplicationDependencies{ + State: state, Operations: staticOperationalStatusReader{}, Configuration: staticConfigurationLoader{}, + Publisher: &recordingConfigurationPublisher{}, + }, options) + if err != nil { + t.Fatalf("NewApplicationService() error = %v", err) + } + return service +} + +type recordingAdminState struct { + mutation adminstate.MutationResult + err error + snapshot adminstate.Snapshot + lastUpstream adminstate.SetUpstreamCommand + lastSwitch adminstate.SwitchRoutingCommand + lastConfig adminstate.CommitConfigCommand + onCommit func(adminstate.CommitConfigCommand) +} + +func (state *recordingAdminState) SetUpstreamEnabled(_ context.Context, command adminstate.SetUpstreamCommand) (adminstate.MutationResult, error) { + state.lastUpstream = command + return state.mutation, state.err +} + +func (state *recordingAdminState) SwitchRouting(_ context.Context, command adminstate.SwitchRoutingCommand) (adminstate.MutationResult, error) { + state.lastSwitch = command + return state.mutation, state.err +} + +func (state *recordingAdminState) CommitConfig(_ context.Context, command adminstate.CommitConfigCommand) (adminstate.MutationResult, error) { + state.lastConfig = command + if state.onCommit != nil { + state.onCommit(command) + } + return state.mutation, state.err +} + +func (state *recordingAdminState) Snapshot(context.Context) (adminstate.Snapshot, error) { + return state.snapshot, state.err +} + +type staticOperationalStatusReader struct { + status OperationalStatus + err error +} + +func (reader staticOperationalStatusReader) ReadOperationalStatus(context.Context) (OperationalStatus, error) { + return reader.status, reader.err +} + +type staticConfigurationLoader struct { + loaded LoadedConfiguration + err error +} + +func (loader staticConfigurationLoader) LoadConfiguration(context.Context) (LoadedConfiguration, error) { + return loader.loaded, loader.err +} + +type recordingConfigurationPublisher struct { + published []*config.Config +} + +func (publisher *recordingConfigurationPublisher) Publish(configuration *config.Config) { + publisher.published = append(publisher.published, configuration) +} diff --git a/progress.md b/progress.md index 6e07e86..052f496 100644 --- a/progress.md +++ b/progress.md @@ -28,6 +28,12 @@ PostgreSQL 18 契约仍待实现。 - Admin Handler 现在复用 `httpsecurity.Identity`,把 Actor ID 与可信 SourceIP 传给 enable/disable/switch/reload mutation,供持久化审计直接使用。 +- 已实现 Admin `ApplicationService`:mutation 统一映射到 `adminstate`;Status 合并 + 一个权威管理快照和低基数运行态聚合;配置重载严格执行解析/校验、脱敏摘要、 + 持久化、原子发布顺序,任何持久化失败都不会替换旧运行配置。 +- 已新增公用原子 `config.Store` 与标准 `FileConfigurationLoader`;覆盖 100 组 + 并发读写、typed-nil、未知字段、主配置/Secret 文件 I/O、取消、Secret 轮换和 + Admin 禁止依赖 Redis Extract/Proxy 明细的架构边界。 - 本轮 `.\scripts\verify.ps1`、`.\scripts\test-redis.ps1`、Compose 静态展开和 Compose 非持久化策略测试通过;Windows `CGO_ENABLED=0`,race 继续由 Linux CI 执行。 diff --git a/task_plan.md b/task_plan.md index 0b1fc5f..55fb277 100644 --- a/task_plan.md +++ b/task_plan.md @@ -29,7 +29,8 @@ 10. [已完成] 实现生产 Redis Activity Adapter、原子 Lua、公用行为契约和 Redis 8.2 集成 fixture;本地 Redis 禁止短效代理数据持久化 11. [进行中] 实现 PostgreSQL 管理面;ADR、领域事务契约、MemoryStore、公用 - 契约、六表 Schema 和 Admin Actor 传播已完成,pgx Adapter/集成测试待完成 + 契约、六表 Schema、Admin Actor 传播、应用服务和原子配置发布已完成,pgx + Adapter/集成测试待完成 ## 串并行关系