feat: connect admin API to management state
This commit is contained in:
youfak
parent
0108b9e813
commit
5b053f0e65
@ -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 无效。
|
||||
|
||||
@ -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 和分配安全余量的批次；已覆盖
|
||||
|
||||
@ -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. 已执行验证
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
@ -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 明细和逐次提取记录。
|
||||
|
||||
43
internal/config/store.go
Normal file
43
internal/config/store.go
Normal file
@ -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)
|
||||
}
|
||||
89
internal/config/store_test.go
Normal file
89
internal/config/store_test.go
Normal file
@ -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"},
|
||||
}},
|
||||
}
|
||||
}
|
||||
84
internal/controller/admin/configuration.go
Normal file
84
internal/controller/admin/configuration.go
Normal file
@ -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
|
||||
}
|
||||
}
|
||||
156
internal/controller/admin/configuration_test.go
Normal file
156
internal/controller/admin/configuration_test.go
Normal file
@ -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}
|
||||
`
|
||||
281
internal/controller/admin/service.go
Normal file
281
internal/controller/admin/service.go
Normal file
@ -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
|
||||
}
|
||||
}
|
||||
43
internal/controller/admin/service_boundary_test.go
Normal file
43
internal/controller/admin/service_boundary_test.go
Normal file
@ -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
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
513
internal/controller/admin/service_test.go
Normal file
513
internal/controller/admin/service_test.go
Normal file
@ -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)
|
||||
}
|
||||
@ -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 执行。
|
||||
|
||||
@ -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/集成测试待完成
|
||||
|
||||
## 串并行关系
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user