diff --git a/README.md b/README.md index 7855de6..d66e84a 100644 --- a/README.md +++ b/README.md @@ -105,7 +105,7 @@ flowchart LR ## 当前完成度 -截至 **2026-08-02**，实施计划中可直接勾选的检查项为 **59 / 74（79.7%）**。详情见 +截至 **2026-08-02**，实施计划中可直接勾选的检查项为 **60 / 74（81.1%）**。详情见 [实施计划](docs/development/implementation-plan.md)和 [交付完成度审计](docs/requirements/completion-audit.md)。 @@ -115,8 +115,8 @@ flowchart LR Redis 会话栅栏，以及 Gateway Outcome 上报的有界队列、序列确认与重试； Controller 的 Redis 共享 BASIC/EGRESS/TARGET 检查任务、按上游的有界轮转调度、HTTP/HTTPS/SOCKS5 Checker 探测和 - Observation 状态归并；Provider 连续空结果的代次化自动 Sequential 切换、禁用候选过滤和 - Snapshot 即时刷新。 + Observation 状态归并；Provider 连续空结果的代次化自动 Sequential 切换、禁用候选过滤、 + 末端 `stop` 的 CAS 路由停用和 Snapshot 即时刷新。 - **部分完成**：Docker Compose/Kubernetes 运行时 mTLS Overlay。 - **待完成**：CONNECT 长连接/Extract 压测场景、故障演练和代表性集群压测。 diff --git a/docs/adr/006-postgresql-admin-state.md b/docs/adr/006-postgresql-admin-state.md index 4687f26..c440846 100644 --- a/docs/adr/006-postgresql-admin-state.md +++ b/docs/adr/006-postgresql-admin-state.md @@ -26,6 +26,7 @@ PostgreSQL 事务内提交。若把它们暴露成多个 Repository，调用方 type Mutator interface { SetUpstreamEnabled(context.Context, SetUpstreamCommand) (MutationResult, error) SwitchRouting(context.Context, SwitchRoutingCommand) (MutationResult, error) + DisableRouting(context.Context, DisableRoutingCommand) (MutationResult, error) CommitConfig(context.Context, CommitConfigCommand) (MutationResult, error) } @@ -43,7 +44,7 @@ type Outbox interface { } ``` -三个 mutation 方法保留现有 Admin 语义，事务、修订号、CAS、审计和 Outbox 都 +四个 mutation 方法保留现有 Admin/自动切换语义，事务、修订号、CAS、审计和 Outbox 都 隐藏在模块实现内部。不会暴露 `BeginTx`、SQL executor 或五个可被错误组合的浅 Repository。 @@ -95,6 +96,12 @@ revision，因此并发提交或 Supervisor 同步的迟到旧版本不能覆盖 并发使用同一 expected 值时最多一个请求成功。目标等于当前值且 expected 匹配时 返回 `changed=false`，仍写审计但不写 Outbox。 +`DisableRouting` 用于 Sequential 的末端 `stop`。它同样锁定目标 Routing，并要求 +`expectedCurrent` 仍等于权威当前值；满足条件时保留 `current_upstream`、仅将 +`enabled` 置为 `false`。这样延迟的 Provider Empty 观察不会覆盖人工切换或配置重载。 +已停用且 expected 匹配时返回 `changed=false` 并审计，不再产生 Outbox；真实停用写入 +`routing.disabled` 事件。该行为与 `SwitchRouting` 共享同一个全局 revision 分配边界。 + ### Outbox 消费 Outbox 使用有界 claim/ack，而不是无界全表扫描： @@ -129,7 +136,8 @@ MemoryStore 与 PostgreSQL Adapter 运行相同契约，至少覆盖： - 配置首次提交、相同重放、校验和冲突和非法引用零写入。 - Upstream enable/disable 幂等、修订单调、审计必写、Outbox 仅在变更时写。 -- Routing CAS、目标校验和 100 个并发请求最多一个成功。 +- Routing CAS、目标校验和 100 个并发请求最多一个成功；terminal stop 的 100 个 + 并发请求仅一个真实停用，其余为幂等审计 no-op。 - 任一审计/Outbox 写故障导致状态完全回滚。 - Outbox 有界 claim、租约到期重试、错误 consumer ACK 拒绝和顺序稳定。 - 审计按 ID 稳定分页并保留 Actor、资源、动作、修订和 UTC 时间；Routing no-op diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index 8249022..6680572 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -325,6 +325,11 @@ Sequential 的空计数属于 Upstream，当前索引属于 Routing。只有 Pro 成功、模板成功且合法候选为零时才增加空计数。错误不改变空计数；重复候选 会重置空计数但增加独立 duplicate 指标。 +Controller 在连续空结果达到阈值时按顺序切换当前 Upstream；到达最后一个可用 +Upstream 且 `endBehavior: stop` 时，会以当前 Upstream 的 CAS 条件原子停用该 +Routing。停用后的 Gateway 规则按既有 `onUnavailable.action` 执行 `reject`、`wait` +或 `direct`，直到一次配置重载提交新的管理快照。 + ## 8. Upstream ```yaml diff --git a/docs/design/architecture.md b/docs/design/architecture.md index 0e6e377..3b05293 100644 --- a/docs/design/architecture.md +++ b/docs/design/architecture.md @@ -319,7 +319,8 @@ Upstream 的 Empty 事实全局共享；每条 Routing 独立 CAS 当前索引 协程只能有一个成功从 A 切到 B，其他协程读取新版本，不会再切到 C。 Sequential 至少配置两个 Upstream；列表耗尽后的默认行为是 `stop`，`loop` 和 `stayLast` 必须显式配置。disabled Upstream 不参与新分配，其运行时跳过与权威 -游标持久化仍由后续 Routing Runtime 完成。 +游标由 Controller 的权威管理状态持久化。达到末端 `stop` 时，Controller 以观察到的 +当前 Upstream 作为 CAS 栅栏禁用 Routing，并让完整 Snapshot 显式关闭该规则。 ## 10. Exclusive Extraction diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 15403cc..fbcc5ee 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -101,7 +101,7 @@ Worker 维度；该测试纳入 Linux CI 的 race 范围，本机因 `CGO_ENABLE - [x] Implement random, round-robin, weighted, least-connections, and sequential. - [x] Model upstream empty counters separately from per-routing current indexes. - [x] Implement versioned CAS switch so simultaneous threshold observers advance once. -- [ ] Cover four-empty-then-success, five-empty, A-to-B-only, disabled references, +- [x] Cover four-empty-then-success, five-empty, A-to-B-only, disabled references, end behavior, and explicit onUnavailable. 当前进度（2026-08-02）：领域构造器与严格配置已统一 Sequential 至少两个 @@ -109,8 +109,10 @@ Upstream、`endBehavior` 默认 `stop`，并覆盖列表末端停止。Provider 连续空结果分配单调代次；Controller 的公共 Sequential 协调器只接收有界通知，在独立 循环中读取权威配置、管理快照和 Stats，并以既有 `ExpectedCurrent` CAS 自动切换。 它会跳过禁用 Upstream，支持 `loop`/`stayLast`，同一空结果代次不会在循环后重复切换， -成功后立即广播完整 Snapshot。Gateway 已执行 reject/wait/direct；控制面将末端 `stop` -持久化为禁用路由的原子管理态变更仍待实现。 +成功后立即广播完整 Snapshot。末端 `stop` 使用独立 `DisableRouting` 公用命令， +以 `ExpectedCurrent` CAS 原子停用 Routing，并在同一权威事务写入审计与 Outbox； +内存与 PostgreSQL Adapter 都运行相同并发、幂等和失败回滚契约。Gateway 已执行 +reject/wait/direct，并会在快照刷新后看到停用状态。 ## Task 5: Provider Fetch Classification and Scheduling diff --git a/docs/operations/production-readiness.md b/docs/operations/production-readiness.md index 4718d0d..5d7b406 100644 --- a/docs/operations/production-readiness.md +++ b/docs/operations/production-readiness.md @@ -5,7 +5,8 @@ - [ ] Gateway 热路径依赖审计确认无 PostgreSQL、Redis、Provider 或模板执行。 - [ ] 每个 Proxy 同一时刻最多归属一个 Worker，ownership epoch 单调。 - [ ] Reserved -> Active 使用单个原子转换，无超卖与负计数。 -- [ ] Sequential 并发 Empty 只切换一次，旧 Upstream Proxy 自然耗尽。 +- [x] Sequential 并发 Empty 只切换一次，旧 Upstream Proxy 自然耗尽；末端 `stop` + 使用权威 CAS 停用 Routing 并刷新 Gateway 快照。 - [x] `pool.maxSize` 与 `fetch.maxTotal` 分别按当前库存和累计获取计数。 - [ ] Extract 只有 `AVAILABLE -> EXTRACTED`，OpenAPI 不存在 release/renew。 - [ ] Extract 状态更新和短期幂等结果位于同一个 Redis 原子操作。 diff --git a/docs/operations/runbook.md b/docs/operations/runbook.md index c3067f4..602a4fe 100644 --- a/docs/operations/runbook.md +++ b/docs/operations/runbook.md @@ -270,7 +270,9 @@ Outbox 发布器必须以稳定 consumer ID 有界领取；发布成功后原子 2. 429 尊重 `Retry-After`，其余 Error 使用指数退避和 jitter。 3. Error 不增加 Empty；合法候选为零才增加 Empty。 4. Duplicate-only 重置 Empty 并记录独立指标。 -5. 达到 Empty 阈值后每条受影响 Routing 只原子切换一次。 +5. 达到 Empty 阈值后每条受影响 Routing 只原子切换一次；最后一个可用 Upstream + 仍持续 Empty 且配置为 `stop` 时，确认审计中存在 `disable_routing` 和 + `routing.disabled` Outbox 事件，并检查 Gateway 已收到关闭该 Routing 的完整快照。 ### 7.5 Gateway 容量耗尽 diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index eebfb6a..a1dbd9e 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -29,8 +29,9 @@ - `PROXY-* / CAP-*`：唯一键、TTL 优先级、状态迁移与 Active/Reserved 打包 原子计数；1,000 goroutine 不超卖，以及 Cancel、重复终结、错误顺序、并发 Commit/Cancel/Release 计数守恒测试。 -- `ROUTE-001 / ROUTE-004`：首条命中规则与进程内 Concurrent Sequential 单次 - 切换；策略运行时接线、持久化恢复和跨实例 CAS 尚未完成。 +- `ROUTE-001 / ROUTE-004`：首条命中规则、Provider 空结果驱动的 Controller + Sequential 切换及末端 stop 已接入权威管理状态；跨 Controller 通过 PostgreSQL + `ExpectedCurrent` CAS 收敛，切换/停用均刷新本进程 Worker Snapshot。 - `FETCH-005 / FETCH-006`：Valid、Empty、DuplicateOnly、Error 分类。 - `DIST-001..003 / DIST-006..007`：内存活动池参考实现验证独占提取、满足模式、 TTL、健康时效与 Gateway 保留量；1,000 并发不重复。 @@ -124,8 +125,7 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 生产任务调度与 REMOVE 编排仍待实现。 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 -10. 将 reject/wait/direct 接入 Distribution 运行链；Sequential 已接入 Provider 空结果 - 代次、管理库 CAS 和禁用候选过滤，仍需补齐末端 `stop` 的原子路由停用管理态。 +10. 将 reject/wait/direct 接入 Distribution 运行链。 ## 4. 容量结论 100,000 QPS 是集群设计输入，不是本次验证结果。只有实现上述运行时，并在 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index 8a0008a..b806593 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -20,7 +20,7 @@ | ROUTE-001 | Routing 自上而下匹配，首条命中停止 | 3534-3798, 5825-6467 | `rule.go` 与不可变/首命中单测 | | ROUTE-002 | Routing 与 Upstream 生命周期解耦 | 3534-3798 | 包依赖与配置模型 | | ROUTE-003 | 支持 sequential、random、roundRobin、weighted、leastConnections | 5825-6467 | 五种领域策略、同版本 Gateway Snapshot 派发和定向测试已完成；Distribution 接线待完成 | -| ROUTE-004 | Sequential 连续空结果达到阈值后原子切换一次 | 5295-5824, 6520-6617 | 进程内 `RoutingCursor` 版本 CAS 与 100 并发测试已完成；Provider Stats 对连续空结果分配单调代次，Controller `SequentialCoordinator` 在独立有界循环中读取权威配置/管理快照，并通过管理库 `ExpectedCurrent` CAS 自动切换。禁用候选会跳过，重复 Tick、循环后旧代次和并发 Tick 均不会再次切换；成功即广播完整 Snapshot | +| ROUTE-004 | Sequential 连续空结果达到阈值后原子切换一次 | 5295-5824, 6520-6617 | 进程内 `RoutingCursor` 版本 CAS 与 100 并发测试已完成；Provider Stats 对连续空结果分配单调代次，Controller `SequentialCoordinator` 在独立有界循环中读取权威配置/管理快照，并通过 `ExpectedCurrent` CAS 自动切换。禁用候选会跳过，重复 Tick、循环后旧代次和并发 Tick 均不会再次切换；末端 `stop` 使用 `DisableRouting` 的同一 CAS 原子停用路由，并与审计/Outbox 同事务提交；两种成功动作都会广播完整 Snapshot | | ROUTE-005 | 空计数属于 Upstream，当前选择属于 Routing | 8442-8529 | 共享 `UpstreamEmptyState` 双 Routing 测试 | | ROUTE-006 | 旧 Upstream 已有 Proxy 继续耗尽，不因切换直接丢弃 | 6618-6641 | Routing 成功切换后立即发布完整快照；Sequential 仅将新分配切到新的 CurrentUpstream，旧 Proxy 仍保留在快照，既有 Active/Reserved 由本地运行态自然归零。共享 Upstream 不按单 Routing 强制 Drain，避免影响其他 Routing | | ROUTE-007 | 无可用 Upstream 时显式 reject、wait 或 direct，默认 reject | 5075-5294, 6743-6760 | Gateway 已实现 reject、带 `wait_timeout` 的本地容量等待与经 TargetPolicy 的 direct；Distribution 接线和默认化策略待完成 | diff --git a/internal/adapters/postgresadmin/adapter_external_test.go b/internal/adapters/postgresadmin/adapter_external_test.go index d581974..0c36f6e 100644 --- a/internal/adapters/postgresadmin/adapter_external_test.go +++ b/internal/adapters/postgresadmin/adapter_external_test.go @@ -65,6 +65,10 @@ func TestOperationsPrioritizeCanceledContext(t *testing.T) { _, operationErr := store.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{}) return operationErr }}, + {name: "disable routing", run: func() error { + _, operationErr := store.DisableRouting(ctx, adminstate.DisableRoutingCommand{}) + return operationErr + }}, {name: "snapshot", run: func() error { _, operationErr := store.Snapshot(ctx) return operationErr @@ -119,6 +123,10 @@ func TestInvalidOperationsDoNotBeginTransaction(t *testing.T) { _, operationErr := store.SwitchRouting(context.Background(), adminstate.SwitchRoutingCommand{}) return operationErr }}, + {name: "disable routing", run: func() error { + _, operationErr := store.DisableRouting(context.Background(), adminstate.DisableRoutingCommand{}) + return operationErr + }}, {name: "read audit", run: func() error { _, operationErr := store.ReadAudit(context.Background(), adminstate.AuditQuery{}) return operationErr diff --git a/internal/adapters/postgresadmin/migrations/0002_disable_routing.sql b/internal/adapters/postgresadmin/migrations/0002_disable_routing.sql new file mode 100644 index 0000000..1e4fc11 --- /dev/null +++ b/internal/adapters/postgresadmin/migrations/0002_disable_routing.sql @@ -0,0 +1,15 @@ +BEGIN; + +ALTER TABLE admin_audit_log + DROP CONSTRAINT IF EXISTS admin_audit_log_action_check; + +ALTER TABLE admin_audit_log + ADD CONSTRAINT admin_audit_log_action_check + CHECK (action IN ( + 'commit_config', + 'set_upstream_enabled', + 'switch_routing', + 'disable_routing' + )); + +COMMIT; diff --git a/internal/adapters/postgresadmin/migrations_test.go b/internal/adapters/postgresadmin/migrations_test.go index b59fc63..f5dc5db 100644 --- a/internal/adapters/postgresadmin/migrations_test.go +++ b/internal/adapters/postgresadmin/migrations_test.go @@ -27,15 +27,18 @@ func TestApplyMigrationsValidatesContextAndPool(t *testing.T) { func TestMigrationsAreOrderedTransactionalAndImmutable(t *testing.T) { t.Parallel() migrations := Migrations() - if len(migrations) != 1 { - t.Fatalf("len(Migrations()) = %d, want 1", len(migrations)) + if len(migrations) != 2 { + t.Fatalf("len(Migrations()) = %d, want 2", len(migrations)) } - if migrations[0].Version != 1 || migrations[0].Name != "admin_state" { - t.Fatalf("migration metadata = %+v", migrations[0]) + if migrations[0].Version != 1 || migrations[0].Name != "admin_state" || + migrations[1].Version != 2 || migrations[1].Name != "disable_routing" { + t.Fatalf("migration metadata = %+v", migrations) } - normalized := strings.TrimSpace(migrations[0].SQL) - if !strings.HasPrefix(normalized, "BEGIN;") || !strings.HasSuffix(normalized, "COMMIT;") { - t.Fatalf("migration is not transaction wrapped: %q", normalized) + for _, migration := range migrations { + normalized := strings.TrimSpace(migration.SQL) + if !strings.HasPrefix(normalized, "BEGIN;") || !strings.HasSuffix(normalized, "COMMIT;") { + t.Fatalf("migration %d is not transaction wrapped: %q", migration.Version, normalized) + } } for index := 1; index < len(migrations); index++ { if migrations[index-1].Version >= migrations[index].Version { @@ -49,6 +52,25 @@ func TestMigrationsAreOrderedTransactionalAndImmutable(t *testing.T) { } } +func TestRoutingDisableMigrationExtendsOnlyAuditActionConstraint(t *testing.T) { + t.Parallel() + lower := strings.ToLower(Migrations()[1].SQL) + for _, required := range []string{ + "alter table admin_audit_log", + "drop constraint if exists admin_audit_log_action_check", + "disable_routing", + } { + if !strings.Contains(lower, required) { + t.Errorf("routing disable migration missing %q", required) + } + } + for _, forbidden := range []string{"create table", "proxy", "credential", "extraction", "ownership", "idempotency"} { + if strings.Contains(lower, forbidden) { + t.Errorf("routing disable migration contains forbidden identifier %q", forbidden) + } + } +} + func TestAdminSchemaContainsOnlyManagementTables(t *testing.T) { t.Parallel() sql := Migrations()[0].SQL diff --git a/internal/adapters/postgresadmin/mutate.go b/internal/adapters/postgresadmin/mutate.go index 26a3578..fd657a4 100644 --- a/internal/adapters/postgresadmin/mutate.go +++ b/internal/adapters/postgresadmin/mutate.go @@ -301,6 +301,96 @@ func (adapter *adapter) SwitchRouting( }, nil } +func (adapter *adapter) DisableRouting( + ctx context.Context, + command adminstate.DisableRoutingCommand, +) (adminstate.MutationResult, error) { + result := adminstate.MutationResult{RequestID: command.RequestID} + if err := contextError(ctx); err != nil { + return result, err + } + if command.Validate() != nil || !adapter.valid() { + return result, adminstate.ErrInvalidCommand + } + + tx, currentRevision, err := adapter.beginMutation(ctx, "disable routing") + if err != nil { + return result, err + } + defer rollback(tx) + + var enabled bool + var current string + if err := tx.QueryRow(ctx, ` + SELECT enabled, current_upstream + FROM routing_admin_state + WHERE name = $1 + FOR UPDATE`, command.Name).Scan(&enabled, ¤t); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return result, adminstate.ErrNotFound + } + return result, databaseError(ctx, "read routing state", err) + } + if current != command.ExpectedCurrent { + return result, adminstate.ErrConflict + } + if !enabled { + if err := insertAudit(ctx, tx, command.RequestID, command.Actor, + adminstate.ActionDisableRoute, "routing", command.Name, false, + currentRevision, command.Reason, command.OccurredAt, "audit routing disable no-op"); err != nil { + return result, err + } + if err := commit(ctx, tx, "commit routing disable no-op"); err != nil { + return result, err + } + result.Revision = uint64(currentRevision) + return result, nil + } + + nextRevision, err := allocateRevision(ctx, tx, currentRevision, "routing", command.OccurredAt) + if err != nil { + return result, err + } + commandTag, err := tx.Exec(ctx, ` + UPDATE routing_admin_state + SET enabled = FALSE, revision = $1, updated_at = $2 + WHERE name = $3 AND enabled = TRUE AND current_upstream = $4`, + nextRevision, utc(command.OccurredAt), command.Name, command.ExpectedCurrent) + if err != nil { + return result, databaseError(ctx, "disable routing state", err) + } + if commandTag.RowsAffected() != 1 { + return result, adminstate.ErrConflict + } + if err := insertAudit(ctx, tx, command.RequestID, command.Actor, + adminstate.ActionDisableRoute, "routing", command.Name, true, + nextRevision, command.Reason, command.OccurredAt, "audit routing disable"); err != nil { + return result, err + } + payload, err := encodePayload(map[string]any{ + "current": command.ExpectedCurrent, + "enabled": false, + "name": command.Name, + "reason": command.Reason, + "revision": nextRevision, + }) + if err != nil { + return result, unavailable("encode routing disable event") + } + if err := insertOutbox(ctx, tx, nextRevision, "routing.disabled", "routing", + command.Name, payload, command.OccurredAt, "insert routing disable event"); err != nil { + return result, err + } + if err := commit(ctx, tx, "commit routing disable"); err != nil { + return result, err + } + return adminstate.MutationResult{ + RequestID: command.RequestID, + Changed: true, + Revision: uint64(nextRevision), + }, nil +} + func (adapter *adapter) beginMutation(ctx context.Context, operation string) (pgx.Tx, int64, error) { tx, err := adapter.begin(ctx, pgx.TxOptions{AccessMode: pgx.ReadWrite}, operation) if err != nil { diff --git a/internal/adapters/postgresadmin/rollback_integration_test.go b/internal/adapters/postgresadmin/rollback_integration_test.go index 8adb580..7506fb0 100644 --- a/internal/adapters/postgresadmin/rollback_integration_test.go +++ b/internal/adapters/postgresadmin/rollback_integration_test.go @@ -62,6 +62,41 @@ func TestMutationRollsBackWhenOutboxInsertFails(t *testing.T) { } } +func TestDisableRoutingRollsBackWhenAuditOrOutboxInsertFails(t *testing.T) { + for _, target := range []struct { + name string + table string + }{ + {name: "audit", table: "admin_audit_log"}, + {name: "outbox", table: "admin_outbox"}, + } { + t.Run(target.name, func(t *testing.T) { + fixture := newPostgresTestFixture(t) + now := integrationNow() + commitIntegrationConfig(t, fixture.Store, now) + installRejectInsertTrigger(t, fixture, target.table, "reject_routing_disable") + + _, err := fixture.Store.DisableRouting(context.Background(), adminstate.DisableRoutingCommand{ + RequestID: "req-routing-disable-failure", Actor: integrationActor(), OccurredAt: now.Add(time.Second), + Name: "checkout", ExpectedCurrent: "provider-a", Reason: "terminal empty results", + }) + if !errors.Is(err, adminstate.ErrUnavailable) { + t.Fatalf("DisableRouting(%s failure) error = %v, want ErrUnavailable", target.name, err) + } + assertFailedRoutingDisableLeftBaseline(t, fixture) + + dropRejectInsertTrigger(t, fixture, target.table, "reject_routing_disable") + result, err := fixture.Store.DisableRouting(context.Background(), adminstate.DisableRoutingCommand{ + RequestID: "req-routing-disable-after-failure", Actor: integrationActor(), OccurredAt: now.Add(2 * time.Second), + Name: "checkout", ExpectedCurrent: "provider-a", Reason: "terminal empty results", + }) + if err != nil || !result.Changed || result.Revision != 2 { + t.Fatalf("DisableRouting(after rollback) = %+v, %v", result, err) + } + }) + } +} + func assertFailedMutationLeftBaseline(t *testing.T, fixture postgresTestFixture) { t.Helper() snapshot, err := fixture.Store.Snapshot(context.Background()) @@ -83,6 +118,28 @@ func assertFailedMutationLeftBaseline(t *testing.T, fixture postgresTestFixture) } } +func assertFailedRoutingDisableLeftBaseline(t *testing.T, fixture postgresTestFixture) { + t.Helper() + snapshot, err := fixture.Store.Snapshot(context.Background()) + if err != nil || snapshot.Revision != 1 || len(snapshot.Routings) != 1 || !snapshot.Routings[0].Enabled || + snapshot.Routings[0].CurrentUpstream != "provider-a" { + t.Fatalf("Snapshot(after failed DisableRouting) = %+v, %v", snapshot, err) + } + audits, err := fixture.Store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 10}) + if err != nil || len(audits) != 1 { + t.Fatalf("ReadAudit(after failed DisableRouting) = %+v, %v", audits, err) + } + for table, want := range map[string]int{"control_revisions": 1, "admin_outbox": 1} { + var count int + if err := fixture.Pool.QueryRow(context.Background(), "SELECT COUNT(*) FROM "+table).Scan(&count); err != nil { + t.Fatalf("count %s: %v", table, err) + } + if count != want { + t.Fatalf("%s row count = %d, want %d", table, count, want) + } + } +} + func installRejectInsertTrigger(t *testing.T, fixture postgresTestFixture, table, trigger string) { t.Helper() function := trigger + "_fn" diff --git a/internal/controller/admin/service_test.go b/internal/controller/admin/service_test.go index 0ba0c3a..380bcc7 100644 --- a/internal/controller/admin/service_test.go +++ b/internal/controller/admin/service_test.go @@ -647,6 +647,7 @@ type recordingAdminState struct { snapshot adminstate.Snapshot lastUpstream adminstate.SetUpstreamCommand lastSwitch adminstate.SwitchRoutingCommand + lastDisable adminstate.DisableRoutingCommand lastConfig adminstate.CommitConfigCommand onCommit func(adminstate.CommitConfigCommand) } @@ -665,6 +666,10 @@ func (state *orderedCommitState) SwitchRouting(context.Context, adminstate.Switc return adminstate.MutationResult{}, nil } +func (state *orderedCommitState) DisableRouting(context.Context, adminstate.DisableRoutingCommand) (adminstate.MutationResult, error) { + return adminstate.MutationResult{}, nil +} + func (state *orderedCommitState) CommitConfig(_ context.Context, command adminstate.CommitConfigCommand) (adminstate.MutationResult, error) { revision := state.next.Add(1) if revision == 1 { @@ -688,6 +693,11 @@ func (state *recordingAdminState) SwitchRouting(_ context.Context, command admin return state.mutation, state.err } +func (state *recordingAdminState) DisableRouting(_ context.Context, command adminstate.DisableRoutingCommand) (adminstate.MutationResult, error) { + state.lastDisable = 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 { diff --git a/internal/controller/routing/sequential.go b/internal/controller/routing/sequential.go index e2b9fd5..0ce198d 100644 --- a/internal/controller/routing/sequential.go +++ b/internal/controller/routing/sequential.go @@ -25,6 +25,7 @@ const ( defaultPollInterval = time.Second autoSwitchActor = "proxy-controller" autoSwitchReason = "consecutive empty provider fetches reached routing threshold" + autoStopReason = "consecutive empty provider fetches reached terminal sequential end" ) // ConfigurationSource supplies one immutable configuration and its matching @@ -38,6 +39,7 @@ type ConfigurationSource interface { type StateStore interface { Snapshot(context.Context) (adminstate.Snapshot, error) SwitchRouting(context.Context, adminstate.SwitchRoutingCommand) (adminstate.MutationResult, error) + DisableRouting(context.Context, adminstate.DisableRoutingCommand) (adminstate.MutationResult, error) } type SnapshotRefreshNotifier interface { @@ -51,6 +53,7 @@ type Options struct { type TickResult struct { Switched int + Stopped int } // SequentialCoordinator observes Provider fetch outcomes without blocking the @@ -179,15 +182,37 @@ func (coordinator *SequentialCoordinator) Tick(ctx context.Context) (TickResult, if coordinator.processed[key] == stats.EmptyGeneration { continue } - target, switchable := nextUpstream(route, state.CurrentUpstream, upstreams, configuration) - if !switchable { + transition := nextTransition(route, state.CurrentUpstream, upstreams, configuration) + if transition.stop { + mutation, mutateErr := coordinator.state.DisableRouting(ctx, adminstate.DisableRoutingCommand{ + RequestID: requestID("stop", route.Name, state.CurrentUpstream, stats.EmptyGeneration), + Actor: adminstate.Actor{ID: autoSwitchActor}, OccurredAt: coordinator.now().UTC(), + Name: route.Name, ExpectedCurrent: state.CurrentUpstream, Reason: autoStopReason, + }) + if mutateErr != nil { + if errors.Is(mutateErr, adminstate.ErrConflict) { + coordinator.processed[key] = stats.EmptyGeneration + continue + } + return result, mutateErr + } + coordinator.processed[key] = stats.EmptyGeneration + if mutation.Changed { + result.Stopped++ + if coordinator.refresh != nil { + coordinator.refresh.NotifySnapshotRefresh() + } + } + continue + } + if transition.target == "" { coordinator.processed[key] = stats.EmptyGeneration continue } mutation, mutateErr := coordinator.state.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{ - RequestID: requestID(route.Name, state.CurrentUpstream, stats.EmptyGeneration), + RequestID: requestID("switch", route.Name, state.CurrentUpstream, stats.EmptyGeneration), Actor: adminstate.Actor{ID: autoSwitchActor}, OccurredAt: coordinator.now().UTC(), - Name: route.Name, ExpectedCurrent: state.CurrentUpstream, Target: target, Reason: autoSwitchReason, + Name: route.Name, ExpectedCurrent: state.CurrentUpstream, Target: transition.target, Reason: autoSwitchReason, }) if mutateErr != nil { if errors.Is(mutateErr, adminstate.ErrConflict) { @@ -207,12 +232,17 @@ func (coordinator *SequentialCoordinator) Tick(ctx context.Context) (TickResult, return result, nil } -func nextUpstream( +type sequentialTransition struct { + target string + stop bool +} + +func nextTransition( route config.Routing, current string, states map[string]adminstate.UpstreamState, configuration *config.Config, -) (string, bool) { +) sequentialTransition { eligible := make([]string, 0, len(route.Upstreams)) currentIndex := -1 for _, upstream := range route.Upstreams { @@ -224,20 +254,25 @@ func nextUpstream( } eligible = append(eligible, upstream) } - if currentIndex < 0 || len(eligible) < 2 { - return "", false + if currentIndex < 0 { + return sequentialTransition{} } if currentIndex+1 < len(eligible) { - return eligible[currentIndex+1], true + return sequentialTransition{target: eligible[currentIndex+1]} } switch route.Strategy.EndBehavior { case "loop": - return eligible[0], eligible[0] != current - case "stayLast", "", "stop": - return "", false + if eligible[0] != current { + return sequentialTransition{target: eligible[0]} + } + case "stop", "": + return sequentialTransition{stop: true} + case "stayLast": + return sequentialTransition{} default: - return "", false + return sequentialTransition{} } + return sequentialTransition{} } func upstreamStates(values []adminstate.UpstreamState) map[string]adminstate.UpstreamState { @@ -273,8 +308,8 @@ func processedKey(routeName, upstream string) string { return routeName + "\x00" + upstream } -func requestID(routeName, upstream string, generation uint64) string { - payload := routeName + "\x00" + upstream + "\x00" + strconv.FormatUint(generation, 10) +func requestID(action, routeName, upstream string, generation uint64) string { + payload := action + "\x00" + routeName + "\x00" + upstream + "\x00" + strconv.FormatUint(generation, 10) digest := sha256.Sum256([]byte(payload)) return "auto-sequential-" + hex.EncodeToString(digest[:16]) } diff --git a/internal/controller/routing/sequential_test.go b/internal/controller/routing/sequential_test.go index 1958af9..7c69c6e 100644 --- a/internal/controller/routing/sequential_test.go +++ b/internal/controller/routing/sequential_test.go @@ -31,6 +31,34 @@ func TestSequentialCoordinatorSwitchesOnlyAtConfiguredThreshold(t *testing.T) { } } +func TestSequentialCoordinatorStopsAtTerminalUpstream(t *testing.T) { + state := newRoutingState("provider-b", map[string]bool{"provider-a": true, "provider-b": true}) + stats := &routingStats{byUpstream: map[string]provider.Stats{ + "provider-b": {UpstreamID: "provider-b", ConsecutiveEmptyFetch: 5, EmptyGeneration: 9}, + }} + refresh := &refreshRecorder{} + coordinator := newCoordinator(t, routingConfiguration("stop", []string{"provider-a", "provider-b"}), state, stats, refresh) + + if result, err := coordinator.Tick(context.Background()); err != nil || result != (TickResult{Stopped: 1}) || state.enabled() || refresh.count != 1 { + t.Fatalf("Tick(terminal stop) = (%+v, %v), enabled=%v refresh=%d", result, err, state.enabled(), refresh.count) + } + if result, err := coordinator.Tick(context.Background()); err != nil || result != (TickResult{}) || state.stops != 1 || refresh.count != 1 { + t.Fatalf("Tick(reused terminal generation) = (%+v, %v), stops=%d refresh=%d", result, err, state.stops, refresh.count) + } +} + +func TestSequentialCoordinatorStopsWhenDisabledCandidatesLeaveNoAlternative(t *testing.T) { + state := newRoutingState("provider-a", map[string]bool{"provider-a": true, "provider-b": false}) + stats := &routingStats{byUpstream: map[string]provider.Stats{ + "provider-a": {UpstreamID: "provider-a", ConsecutiveEmptyFetch: 5, EmptyGeneration: 4}, + }} + coordinator := newCoordinator(t, routingConfiguration("stop", []string{"provider-a", "provider-b"}), state, stats, &refreshRecorder{}) + + if result, err := coordinator.Tick(context.Background()); err != nil || result != (TickResult{Stopped: 1}) || state.enabled() { + t.Fatalf("Tick(last enabled candidate) = (%+v, %v), enabled=%v", result, err, state.enabled()) + } +} + func TestSequentialCoordinatorConsumesStatsRecorderNotifications(t *testing.T) { state := newRoutingState("provider-a", map[string]bool{"provider-a": true, "provider-b": true}) stats, err := provider.NewStatsRecorder(2) @@ -206,6 +234,7 @@ type routingState struct { mu sync.Mutex snapshot adminstate.Snapshot switches int + stops int } func newRoutingState(current string, enabled map[string]bool) *routingState { @@ -237,12 +266,33 @@ func (state *routingState) SwitchRouting(_ context.Context, command adminstate.S return adminstate.MutationResult{RequestID: command.RequestID, Changed: true, Revision: uint64(state.switches)}, nil } +func (state *routingState) DisableRouting(_ context.Context, command adminstate.DisableRoutingCommand) (adminstate.MutationResult, error) { + state.mu.Lock() + defer state.mu.Unlock() + routing := &state.snapshot.Routings[0] + if command.Name != routing.Name || command.ExpectedCurrent != routing.CurrentUpstream { + return adminstate.MutationResult{}, adminstate.ErrConflict + } + if !routing.Enabled { + return adminstate.MutationResult{RequestID: command.RequestID}, nil + } + routing.Enabled = false + state.stops++ + return adminstate.MutationResult{RequestID: command.RequestID, Changed: true, Revision: uint64(state.stops)}, nil +} + func (state *routingState) current() string { state.mu.Lock() defer state.mu.Unlock() return state.snapshot.Routings[0].CurrentUpstream } +func (state *routingState) enabled() bool { + state.mu.Lock() + defer state.mu.Unlock() + return state.snapshot.Routings[0].Enabled +} + type routingStats struct { mu sync.Mutex byUpstream map[string]provider.Stats diff --git a/internal/domain/adminstate/adminstate.go b/internal/domain/adminstate/adminstate.go index 45f8afe..23e5808 100644 --- a/internal/domain/adminstate/adminstate.go +++ b/internal/domain/adminstate/adminstate.go @@ -35,6 +35,7 @@ var identifierPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$`) type Mutator interface { SetUpstreamEnabled(context.Context, SetUpstreamCommand) (MutationResult, error) SwitchRouting(context.Context, SwitchRoutingCommand) (MutationResult, error) + DisableRouting(context.Context, DisableRoutingCommand) (MutationResult, error) CommitConfig(context.Context, CommitConfigCommand) (MutationResult, error) } @@ -81,6 +82,18 @@ type SwitchRoutingCommand struct { Reason string } +// DisableRoutingCommand atomically stops a Routing only when it still points +// at the upstream observed by the caller. The expected-current fence prevents +// delayed Provider observations from overriding a newer manual switch. +type DisableRoutingCommand struct { + RequestID string + Actor Actor + OccurredAt time.Time + Name string + ExpectedCurrent string + Reason string +} + type CommitConfigCommand struct { RequestID string Actor Actor @@ -147,6 +160,7 @@ type Action string const ( ActionSetUpstream Action = "set_upstream_enabled" ActionSwitchRoute Action = "switch_routing" + ActionDisableRoute Action = "disable_routing" ActionCommitConfig Action = "commit_config" ) @@ -213,6 +227,17 @@ func (command SwitchRoutingCommand) Validate() error { return nil } +// Validate checks whether the terminal Routing stop command is safe for every +// Store adapter to execute. +func (command DisableRoutingCommand) Validate() error { + if err := validateMutationBase(command.RequestID, command.Actor, command.OccurredAt); err != nil || + !validIdentifier(command.Name) || !validIdentifier(command.ExpectedCurrent) || + !validOptionalText(command.Reason, MaxReasonBytes) { + return ErrInvalidCommand + } + return nil +} + // Validate checks the complete management snapshot and all of its references. func (command CommitConfigCommand) Validate() error { if err := validateMutationBase(command.RequestID, command.Actor, command.OccurredAt); err != nil || diff --git a/internal/domain/adminstate/contracttest/contract.go b/internal/domain/adminstate/contracttest/contract.go index 8b90ea0..aa781fd 100644 --- a/internal/domain/adminstate/contracttest/contract.go +++ b/internal/domain/adminstate/contracttest/contract.go @@ -29,6 +29,9 @@ func Run(t *testing.T, factory Factory) { t.Run("routing no-op audit", func(t *testing.T) { runRoutingNoOpContract(t, factory(t)) }) + t.Run("routing terminal stop compare and swap", func(t *testing.T) { + runRoutingDisableContract(t, factory(t)) + }) t.Run("audit pagination and fields", func(t *testing.T) { runAuditContract(t, factory(t)) }) @@ -219,6 +222,94 @@ func runRoutingNoOpContract(t *testing.T, store adminstate.Store) { } } +func runRoutingDisableContract(t *testing.T, store adminstate.Store) { + t.Helper() + now := contractNow() + commit(t, store, configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes))) + + start := make(chan struct{}) + var changed atomic.Int64 + var noops atomic.Int64 + var unexpectedMu sync.Mutex + var unexpected []error + var workers sync.WaitGroup + for index := range 100 { + workers.Add(1) + go func(index int) { + defer workers.Done() + <-start + result, err := store.DisableRouting(context.Background(), adminstate.DisableRoutingCommand{ + RequestID: fmt.Sprintf("req-stop-%03d", index), Actor: contractActor(), + OccurredAt: now.Add(time.Second), Name: "checkout", ExpectedCurrent: "provider-a", + Reason: "terminal empty results", + }) + switch { + case err == nil && result.Changed: + changed.Add(1) + case err == nil && !result.Changed && result.Revision == 2: + noops.Add(1) + default: + unexpectedMu.Lock() + unexpected = append(unexpected, err) + unexpectedMu.Unlock() + } + }(index) + } + close(start) + workers.Wait() + if changed.Load() != 1 || noops.Load() != 99 || len(unexpected) != 0 { + t.Fatalf("concurrent DisableRouting changed=%d noops=%d unexpected=%v", changed.Load(), noops.Load(), unexpected) + } + + snapshot, err := store.Snapshot(context.Background()) + if err != nil || snapshot.Revision != 2 || routingEnabled(snapshot, "checkout") || + snapshot.Routings[0].CurrentUpstream != "provider-a" { + t.Fatalf("Snapshot(after DisableRouting) = %+v, %v", snapshot, err) + } + audits, err := store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 200}) + if err != nil || len(audits) != 101 { + t.Fatalf("ReadAudit(after DisableRouting) = %d records, %v", len(audits), err) + } + var changedStops int + for _, audit := range audits { + if audit.Action == adminstate.ActionDisableRoute && audit.Changed { + changedStops++ + if audit.ResourceType != "routing" || audit.ResourceName != "checkout" || + audit.Revision != 2 || audit.Reason != "terminal empty results" { + t.Fatalf("changed disable audit = %+v", audit) + } + } + } + if changedStops != 1 { + t.Fatalf("changed disable audits = %d, want 1", changedStops) + } + events, err := store.Claim(context.Background(), adminstate.ClaimCommand{ + ConsumerID: "publisher-a", Now: now.Add(2 * time.Second), Limit: 10, Lease: time.Minute, + }) + if err != nil || len(events) != 2 || events[1].Type != "routing.disabled" || + events[1].AggregateType != "routing" || events[1].AggregateID != "checkout" || events[1].Revision != 2 { + t.Fatalf("Claim(after DisableRouting) = %+v, %v", events, err) + } + + before := len(audits) + if _, err := store.DisableRouting(context.Background(), adminstate.DisableRoutingCommand{ + RequestID: "req-stale-stop", Actor: contractActor(), OccurredAt: now.Add(3 * time.Second), + Name: "checkout", ExpectedCurrent: "provider-b", + }); !errors.Is(err, adminstate.ErrConflict) { + t.Fatalf("DisableRouting(stale) error = %v, want ErrConflict", err) + } + if _, err := store.DisableRouting(context.Background(), adminstate.DisableRoutingCommand{ + RequestID: "req-missing-stop", Actor: contractActor(), OccurredAt: now.Add(3 * time.Second), + Name: "missing", ExpectedCurrent: "provider-a", + }); !errors.Is(err, adminstate.ErrNotFound) { + t.Fatalf("DisableRouting(missing) error = %v, want ErrNotFound", err) + } + audits, err = store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 200}) + if err != nil || len(audits) != before { + t.Fatalf("ReadAudit(after rejected DisableRouting) = %d records, %v; want %d", len(audits), err, before) + } +} + func runAuditContract(t *testing.T, store adminstate.Store) { t.Helper() now := contractNow() @@ -355,6 +446,12 @@ func runContextContract(t *testing.T, store adminstate.Store) { }); !errors.Is(err, context.Canceled) { t.Fatalf("SwitchRouting(canceled) error = %v", err) } + if _, err := store.DisableRouting(ctx, adminstate.DisableRoutingCommand{ + RequestID: "req-stop", Actor: contractActor(), OccurredAt: contractNow(), + Name: "checkout", ExpectedCurrent: "provider-a", + }); !errors.Is(err, context.Canceled) { + t.Fatalf("DisableRouting(canceled) error = %v", err) + } if _, err := store.Snapshot(ctx); !errors.Is(err, context.Canceled) { t.Fatalf("Snapshot(canceled) error = %v", err) } @@ -411,3 +508,12 @@ func upstreamEnabled(snapshot adminstate.Snapshot, name string) bool { } return false } + +func routingEnabled(snapshot adminstate.Snapshot, name string) bool { + for _, routing := range snapshot.Routings { + if routing.Name == name { + return routing.Enabled + } + } + return false +} diff --git a/internal/domain/adminstate/memory.go b/internal/domain/adminstate/memory.go index 348b06c..45a29f7 100644 --- a/internal/domain/adminstate/memory.go +++ b/internal/domain/adminstate/memory.go @@ -210,6 +210,60 @@ func (store *MemoryStore) SwitchRouting(ctx context.Context, command SwitchRouti return MutationResult{RequestID: command.RequestID, Changed: true, Revision: nextRevision}, nil } +func (store *MemoryStore) DisableRouting(ctx context.Context, command DisableRoutingCommand) (MutationResult, error) { + result := MutationResult{RequestID: command.RequestID} + if err := contextError(ctx); err != nil { + return result, err + } + if store == nil || command.Validate() != nil { + return result, ErrInvalidCommand + } + + store.mu.Lock() + defer store.mu.Unlock() + if err := ctx.Err(); err != nil { + return result, err + } + state, exists := store.routings[command.Name] + if !exists { + return result, ErrNotFound + } + if state.CurrentUpstream != command.ExpectedCurrent { + return result, ErrConflict + } + if !state.Enabled { + result.Revision = store.revision + store.appendAuditLocked(command.RequestID, command.Actor, ActionDisableRoute, "routing", + command.Name, false, store.revision, command.Reason, command.OccurredAt) + return result, nil + } + if store.revision == math.MaxUint64 { + return result, ErrUnavailable + } + + nextRevision := store.revision + 1 + payload, err := encodeEventPayload(map[string]any{ + "current": command.ExpectedCurrent, + "enabled": false, + "name": command.Name, + "reason": command.Reason, + "revision": nextRevision, + }) + if err != nil { + return result, ErrUnavailable + } + state.Enabled = false + state.Revision = nextRevision + state.UpdatedAt = command.OccurredAt.UTC() + store.revision = nextRevision + store.routings[command.Name] = state + store.appendAuditLocked(command.RequestID, command.Actor, ActionDisableRoute, "routing", + command.Name, true, nextRevision, command.Reason, command.OccurredAt) + store.appendEventLocked(nextRevision, "routing.disabled", "routing", command.Name, + payload, command.OccurredAt) + return MutationResult{RequestID: command.RequestID, Changed: true, Revision: nextRevision}, nil +} + func (store *MemoryStore) Snapshot(ctx context.Context) (Snapshot, error) { if err := contextError(ctx); err != nil { return Snapshot{}, err diff --git a/internal/domain/adminstate/validation_test.go b/internal/domain/adminstate/validation_test.go index 694e80d..7e700bc 100644 --- a/internal/domain/adminstate/validation_test.go +++ b/internal/domain/adminstate/validation_test.go @@ -74,6 +74,37 @@ func TestValidateSwitchRoutingCommand(t *testing.T) { } } +func TestValidateDisableRoutingCommand(t *testing.T) { + t.Parallel() + now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) + valid := DisableRoutingCommand{ + RequestID: "req-disable-routing", Actor: Actor{ID: "proxy-controller"}, OccurredAt: now, + Name: "checkout", ExpectedCurrent: "provider-b", Reason: "terminal empty results", + } + if err := valid.Validate(); err != nil { + t.Fatalf("DisableRoutingCommand.Validate(valid): %v", err) + } + + for _, test := range []struct { + name string + mutate func(*DisableRoutingCommand) + }{ + {name: "missing routing", mutate: func(command *DisableRoutingCommand) { command.Name = "" }}, + {name: "missing expected", mutate: func(command *DisableRoutingCommand) { command.ExpectedCurrent = "" }}, + {name: "reason too long", mutate: func(command *DisableRoutingCommand) { command.Reason = strings.Repeat("r", MaxReasonBytes+1) }}, + } { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + command := valid + test.mutate(&command) + if err := command.Validate(); !errors.Is(err, ErrInvalidCommand) { + t.Fatalf("DisableRoutingCommand.Validate() error = %v, want ErrInvalidCommand", err) + } + }) + } +} + func TestValidateCommitConfigCommandAndReferences(t *testing.T) { t.Parallel() valid := validCommitConfigCommand()