feat: persist terminal sequential stops

This commit is contained in:
youfak 2026-08-02 15:14:20 +08:00
parent 589cd4e454
commit e220d5368c
21 changed files with 560 additions and 38 deletions

View File

@ -105,7 +105,7 @@ flowchart LR
## 当前完成度 ## 当前完成度
截至 **2026-08-02**,实施计划中可直接勾选的检查项为 **59 / 7479.7%**。详情见 截至 **2026-08-02**,实施计划中可直接勾选的检查项为 **60 / 7481.1%**。详情见
[实施计划](docs/development/implementation-plan.md)和 [实施计划](docs/development/implementation-plan.md)和
[交付完成度审计](docs/requirements/completion-audit.md)。 [交付完成度审计](docs/requirements/completion-audit.md)。
@ -115,8 +115,8 @@ flowchart LR
Redis 会话栅栏,以及 Gateway Outcome 上报的有界队列、序列确认与重试; Redis 会话栅栏,以及 Gateway Outcome 上报的有界队列、序列确认与重试;
Controller 的 Redis 共享 BASIC/EGRESS/TARGET 检查任务、按上游的有界轮转调度、HTTP/HTTPS/SOCKS5 Controller 的 Redis 共享 BASIC/EGRESS/TARGET 检查任务、按上游的有界轮转调度、HTTP/HTTPS/SOCKS5
Checker 探测和 Checker 探测和
Observation 状态归并Provider 连续空结果的代次化自动 Sequential 切换、禁用候选过滤 Observation 状态归并Provider 连续空结果的代次化自动 Sequential 切换、禁用候选过滤
Snapshot 即时刷新。 末端 `stop` 的 CAS 路由停用和 Snapshot 即时刷新。
- **部分完成**Docker Compose/Kubernetes 运行时 mTLS Overlay。 - **部分完成**Docker Compose/Kubernetes 运行时 mTLS Overlay。
- **待完成**CONNECT 长连接/Extract 压测场景、故障演练和代表性集群压测。 - **待完成**CONNECT 长连接/Extract 压测场景、故障演练和代表性集群压测。

View File

@ -26,6 +26,7 @@ PostgreSQL 事务内提交。若把它们暴露成多个 Repository调用方
type Mutator interface { type Mutator interface {
SetUpstreamEnabled(context.Context, SetUpstreamCommand) (MutationResult, error) SetUpstreamEnabled(context.Context, SetUpstreamCommand) (MutationResult, error)
SwitchRouting(context.Context, SwitchRoutingCommand) (MutationResult, error) SwitchRouting(context.Context, SwitchRoutingCommand) (MutationResult, error)
DisableRouting(context.Context, DisableRoutingCommand) (MutationResult, error)
CommitConfig(context.Context, CommitConfigCommand) (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 或五个可被错误组合的浅 隐藏在模块实现内部。不会暴露 `BeginTx`、SQL executor 或五个可被错误组合的浅
Repository。 Repository。
@ -95,6 +96,12 @@ revision因此并发提交或 Supervisor 同步的迟到旧版本不能覆盖
并发使用同一 expected 值时最多一个请求成功。目标等于当前值且 expected 匹配时 并发使用同一 expected 值时最多一个请求成功。目标等于当前值且 expected 匹配时
返回 `changed=false`,仍写审计但不写 Outbox。 返回 `changed=false`,仍写审计但不写 Outbox。
`DisableRouting` 用于 Sequential 的末端 `stop`。它同样锁定目标 Routing并要求
`expectedCurrent` 仍等于权威当前值;满足条件时保留 `current_upstream`、仅将
`enabled` 置为 `false`。这样延迟的 Provider Empty 观察不会覆盖人工切换或配置重载。
已停用且 expected 匹配时返回 `changed=false` 并审计,不再产生 Outbox真实停用写入
`routing.disabled` 事件。该行为与 `SwitchRouting` 共享同一个全局 revision 分配边界。
### Outbox 消费 ### Outbox 消费
Outbox 使用有界 claim/ack而不是无界全表扫描 Outbox 使用有界 claim/ack而不是无界全表扫描
@ -129,7 +136,8 @@ MemoryStore 与 PostgreSQL Adapter 运行相同契约,至少覆盖:
- 配置首次提交、相同重放、校验和冲突和非法引用零写入。 - 配置首次提交、相同重放、校验和冲突和非法引用零写入。
- Upstream enable/disable 幂等、修订单调、审计必写、Outbox 仅在变更时写。 - Upstream enable/disable 幂等、修订单调、审计必写、Outbox 仅在变更时写。
- Routing CAS、目标校验和 100 个并发请求最多一个成功。 - Routing CAS、目标校验和 100 个并发请求最多一个成功terminal stop 的 100 个
并发请求仅一个真实停用,其余为幂等审计 no-op。
- 任一审计/Outbox 写故障导致状态完全回滚。 - 任一审计/Outbox 写故障导致状态完全回滚。
- Outbox 有界 claim、租约到期重试、错误 consumer ACK 拒绝和顺序稳定。 - Outbox 有界 claim、租约到期重试、错误 consumer ACK 拒绝和顺序稳定。
- 审计按 ID 稳定分页并保留 Actor、资源、动作、修订和 UTC 时间Routing no-op - 审计按 ID 稳定分页并保留 Actor、资源、动作、修订和 UTC 时间Routing no-op

View File

@ -325,6 +325,11 @@ Sequential 的空计数属于 Upstream当前索引属于 Routing。只有 Pro
成功、模板成功且合法候选为零时才增加空计数。错误不改变空计数;重复候选 成功、模板成功且合法候选为零时才增加空计数。错误不改变空计数;重复候选
会重置空计数但增加独立 duplicate 指标。 会重置空计数但增加独立 duplicate 指标。
Controller 在连续空结果达到阈值时按顺序切换当前 Upstream到达最后一个可用
Upstream 且 `endBehavior: stop` 时,会以当前 Upstream 的 CAS 条件原子停用该
Routing。停用后的 Gateway 规则按既有 `onUnavailable.action` 执行 `reject`、`wait`
`direct`,直到一次配置重载提交新的管理快照。
## 8. Upstream ## 8. Upstream
```yaml ```yaml

View File

@ -319,7 +319,8 @@ Upstream 的 Empty 事实全局共享;每条 Routing 独立 CAS 当前索引
协程只能有一个成功从 A 切到 B其他协程读取新版本不会再切到 C。 协程只能有一个成功从 A 切到 B其他协程读取新版本不会再切到 C。
Sequential 至少配置两个 Upstream列表耗尽后的默认行为是 `stop``loop` 和 Sequential 至少配置两个 Upstream列表耗尽后的默认行为是 `stop``loop` 和
`stayLast` 必须显式配置。disabled Upstream 不参与新分配,其运行时跳过与权威 `stayLast` 必须显式配置。disabled Upstream 不参与新分配,其运行时跳过与权威
游标持久化仍由后续 Routing Runtime 完成。 游标由 Controller 的权威管理状态持久化。达到末端 `stop`Controller 以观察到的
当前 Upstream 作为 CAS 栅栏禁用 Routing并让完整 Snapshot 显式关闭该规则。
## 10. Exclusive Extraction ## 10. Exclusive Extraction

View File

@ -101,7 +101,7 @@ Worker 维度;该测试纳入 Linux CI 的 race 范围,本机因 `CGO_ENABLE
- [x] Implement random, round-robin, weighted, least-connections, and sequential. - [x] Implement random, round-robin, weighted, least-connections, and sequential.
- [x] Model upstream empty counters separately from per-routing current indexes. - [x] Model upstream empty counters separately from per-routing current indexes.
- [x] Implement versioned CAS switch so simultaneous threshold observers advance once. - [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. end behavior, and explicit onUnavailable.
当前进度2026-08-02领域构造器与严格配置已统一 Sequential 至少两个 当前进度2026-08-02领域构造器与严格配置已统一 Sequential 至少两个
@ -109,8 +109,10 @@ Upstream、`endBehavior` 默认 `stop`并覆盖列表末端停止。Provider
连续空结果分配单调代次Controller 的公共 Sequential 协调器只接收有界通知,在独立 连续空结果分配单调代次Controller 的公共 Sequential 协调器只接收有界通知,在独立
循环中读取权威配置、管理快照和 Stats并以既有 `ExpectedCurrent` CAS 自动切换。 循环中读取权威配置、管理快照和 Stats并以既有 `ExpectedCurrent` CAS 自动切换。
它会跳过禁用 Upstream支持 `loop`/`stayLast`,同一空结果代次不会在循环后重复切换, 它会跳过禁用 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 ## Task 5: Provider Fetch Classification and Scheduling

View File

@ -5,7 +5,8 @@
- [ ] Gateway 热路径依赖审计确认无 PostgreSQL、Redis、Provider 或模板执行。 - [ ] Gateway 热路径依赖审计确认无 PostgreSQL、Redis、Provider 或模板执行。
- [ ] 每个 Proxy 同一时刻最多归属一个 Workerownership epoch 单调。 - [ ] 每个 Proxy 同一时刻最多归属一个 Workerownership epoch 单调。
- [ ] Reserved -> Active 使用单个原子转换,无超卖与负计数。 - [ ] Reserved -> Active 使用单个原子转换,无超卖与负计数。
- [ ] Sequential 并发 Empty 只切换一次,旧 Upstream Proxy 自然耗尽。 - [x] Sequential 并发 Empty 只切换一次,旧 Upstream Proxy 自然耗尽;末端 `stop`
使用权威 CAS 停用 Routing 并刷新 Gateway 快照。
- [x] `pool.maxSize``fetch.maxTotal` 分别按当前库存和累计获取计数。 - [x] `pool.maxSize``fetch.maxTotal` 分别按当前库存和累计获取计数。
- [ ] Extract 只有 `AVAILABLE -> EXTRACTED`OpenAPI 不存在 release/renew。 - [ ] Extract 只有 `AVAILABLE -> EXTRACTED`OpenAPI 不存在 release/renew。
- [ ] Extract 状态更新和短期幂等结果位于同一个 Redis 原子操作。 - [ ] Extract 状态更新和短期幂等结果位于同一个 Redis 原子操作。

View File

@ -270,7 +270,9 @@ Outbox 发布器必须以稳定 consumer ID 有界领取;发布成功后原子
2. 429 尊重 `Retry-After`,其余 Error 使用指数退避和 jitter。 2. 429 尊重 `Retry-After`,其余 Error 使用指数退避和 jitter。
3. Error 不增加 Empty合法候选为零才增加 Empty。 3. Error 不增加 Empty合法候选为零才增加 Empty。
4. Duplicate-only 重置 Empty 并记录独立指标。 4. Duplicate-only 重置 Empty 并记录独立指标。
5. 达到 Empty 阈值后每条受影响 Routing 只原子切换一次。 5. 达到 Empty 阈值后每条受影响 Routing 只原子切换一次;最后一个可用 Upstream
仍持续 Empty 且配置为 `stop` 时,确认审计中存在 `disable_routing`
`routing.disabled` Outbox 事件,并检查 Gateway 已收到关闭该 Routing 的完整快照。
### 7.5 Gateway 容量耗尽 ### 7.5 Gateway 容量耗尽

View File

@ -29,8 +29,9 @@
- `PROXY-* / CAP-*`唯一键、TTL 优先级、状态迁移与 Active/Reserved 打包 - `PROXY-* / CAP-*`唯一键、TTL 优先级、状态迁移与 Active/Reserved 打包
原子计数1,000 goroutine 不超卖,以及 Cancel、重复终结、错误顺序、并发 原子计数1,000 goroutine 不超卖,以及 Cancel、重复终结、错误顺序、并发
Commit/Cancel/Release 计数守恒测试。 Commit/Cancel/Release 计数守恒测试。
- `ROUTE-001 / ROUTE-004`:首条命中规则与进程内 Concurrent Sequential 单次 - `ROUTE-001 / ROUTE-004`首条命中规则、Provider 空结果驱动的 Controller
切换;策略运行时接线、持久化恢复和跨实例 CAS 尚未完成。 Sequential 切换及末端 stop 已接入权威管理状态;跨 Controller 通过 PostgreSQL
`ExpectedCurrent` CAS 收敛,切换/停用均刷新本进程 Worker Snapshot。
- `FETCH-005 / FETCH-006`Valid、Empty、DuplicateOnly、Error 分类。 - `FETCH-005 / FETCH-006`Valid、Empty、DuplicateOnly、Error 分类。
- `DIST-001..003 / DIST-006..007`:内存活动池参考实现验证独占提取、满足模式、 - `DIST-001..003 / DIST-006..007`:内存活动池参考实现验证独占提取、满足模式、
TTL、健康时效与 Gateway 保留量1,000 并发不重复。 TTL、健康时效与 Gateway 保留量1,000 并发不重复。
@ -124,8 +125,7 @@ Controller/Gateway 入口,完整 mTLS 运行时拓扑仍只有静态验证。
生产任务调度与 REMOVE 编排仍待实现。 生产任务调度与 REMOVE 编排仍待实现。
8. Admin/Distribution 细粒度授权和审计查询Distribution 分布式限流已完成。 8. Admin/Distribution 细粒度授权和审计查询Distribution 分布式限流已完成。
9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。
10. 将 reject/wait/direct 接入 Distribution 运行链Sequential 已接入 Provider 空结果 10. 将 reject/wait/direct 接入 Distribution 运行链。
代次、管理库 CAS 和禁用候选过滤,仍需补齐末端 `stop` 的原子路由停用管理态。
## 4. 容量结论 ## 4. 容量结论
100,000 QPS 是集群设计输入,不是本次验证结果。只有实现上述运行时,并在 100,000 QPS 是集群设计输入,不是本次验证结果。只有实现上述运行时,并在

View File

@ -20,7 +20,7 @@
| ROUTE-001 | Routing 自上而下匹配,首条命中停止 | 3534-3798, 5825-6467 | `rule.go` 与不可变/首命中单测 | | ROUTE-001 | Routing 自上而下匹配,首条命中停止 | 3534-3798, 5825-6467 | `rule.go` 与不可变/首命中单测 |
| ROUTE-002 | Routing 与 Upstream 生命周期解耦 | 3534-3798 | 包依赖与配置模型 | | ROUTE-002 | Routing 与 Upstream 生命周期解耦 | 3534-3798 | 包依赖与配置模型 |
| ROUTE-003 | 支持 sequential、random、roundRobin、weighted、leastConnections | 5825-6467 | 五种领域策略、同版本 Gateway Snapshot 派发和定向测试已完成Distribution 接线待完成 | | 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-005 | 空计数属于 Upstream当前选择属于 Routing | 8442-8529 | 共享 `UpstreamEmptyState` 双 Routing 测试 |
| ROUTE-006 | 旧 Upstream 已有 Proxy 继续耗尽,不因切换直接丢弃 | 6618-6641 | Routing 成功切换后立即发布完整快照Sequential 仅将新分配切到新的 CurrentUpstream旧 Proxy 仍保留在快照,既有 Active/Reserved 由本地运行态自然归零。共享 Upstream 不按单 Routing 强制 Drain避免影响其他 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 的 directDistribution 接线和默认化策略待完成 | | ROUTE-007 | 无可用 Upstream 时显式 reject、wait 或 direct默认 reject | 5075-5294, 6743-6760 | Gateway 已实现 reject、带 `wait_timeout` 的本地容量等待与经 TargetPolicy 的 directDistribution 接线和默认化策略待完成 |

View File

@ -65,6 +65,10 @@ func TestOperationsPrioritizeCanceledContext(t *testing.T) {
_, operationErr := store.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{}) _, operationErr := store.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{})
return operationErr return operationErr
}}, }},
{name: "disable routing", run: func() error {
_, operationErr := store.DisableRouting(ctx, adminstate.DisableRoutingCommand{})
return operationErr
}},
{name: "snapshot", run: func() error { {name: "snapshot", run: func() error {
_, operationErr := store.Snapshot(ctx) _, operationErr := store.Snapshot(ctx)
return operationErr return operationErr
@ -119,6 +123,10 @@ func TestInvalidOperationsDoNotBeginTransaction(t *testing.T) {
_, operationErr := store.SwitchRouting(context.Background(), adminstate.SwitchRoutingCommand{}) _, operationErr := store.SwitchRouting(context.Background(), adminstate.SwitchRoutingCommand{})
return operationErr return operationErr
}}, }},
{name: "disable routing", run: func() error {
_, operationErr := store.DisableRouting(context.Background(), adminstate.DisableRoutingCommand{})
return operationErr
}},
{name: "read audit", run: func() error { {name: "read audit", run: func() error {
_, operationErr := store.ReadAudit(context.Background(), adminstate.AuditQuery{}) _, operationErr := store.ReadAudit(context.Background(), adminstate.AuditQuery{})
return operationErr return operationErr

View File

@ -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;

View File

@ -27,15 +27,18 @@ func TestApplyMigrationsValidatesContextAndPool(t *testing.T) {
func TestMigrationsAreOrderedTransactionalAndImmutable(t *testing.T) { func TestMigrationsAreOrderedTransactionalAndImmutable(t *testing.T) {
t.Parallel() t.Parallel()
migrations := Migrations() migrations := Migrations()
if len(migrations) != 1 { if len(migrations) != 2 {
t.Fatalf("len(Migrations()) = %d, want 1", len(migrations)) t.Fatalf("len(Migrations()) = %d, want 2", len(migrations))
} }
if migrations[0].Version != 1 || migrations[0].Name != "admin_state" { if migrations[0].Version != 1 || migrations[0].Name != "admin_state" ||
t.Fatalf("migration metadata = %+v", migrations[0]) migrations[1].Version != 2 || migrations[1].Name != "disable_routing" {
t.Fatalf("migration metadata = %+v", migrations)
} }
normalized := strings.TrimSpace(migrations[0].SQL) for _, migration := range migrations {
if !strings.HasPrefix(normalized, "BEGIN;") || !strings.HasSuffix(normalized, "COMMIT;") { normalized := strings.TrimSpace(migration.SQL)
t.Fatalf("migration is not transaction wrapped: %q", normalized) 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++ { for index := 1; index < len(migrations); index++ {
if migrations[index-1].Version >= migrations[index].Version { 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) { func TestAdminSchemaContainsOnlyManagementTables(t *testing.T) {
t.Parallel() t.Parallel()
sql := Migrations()[0].SQL sql := Migrations()[0].SQL

View File

@ -301,6 +301,96 @@ func (adapter *adapter) SwitchRouting(
}, nil }, 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, &current); 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) { func (adapter *adapter) beginMutation(ctx context.Context, operation string) (pgx.Tx, int64, error) {
tx, err := adapter.begin(ctx, pgx.TxOptions{AccessMode: pgx.ReadWrite}, operation) tx, err := adapter.begin(ctx, pgx.TxOptions{AccessMode: pgx.ReadWrite}, operation)
if err != nil { if err != nil {

View File

@ -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) { func assertFailedMutationLeftBaseline(t *testing.T, fixture postgresTestFixture) {
t.Helper() t.Helper()
snapshot, err := fixture.Store.Snapshot(context.Background()) 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) { func installRejectInsertTrigger(t *testing.T, fixture postgresTestFixture, table, trigger string) {
t.Helper() t.Helper()
function := trigger + "_fn" function := trigger + "_fn"

View File

@ -647,6 +647,7 @@ type recordingAdminState struct {
snapshot adminstate.Snapshot snapshot adminstate.Snapshot
lastUpstream adminstate.SetUpstreamCommand lastUpstream adminstate.SetUpstreamCommand
lastSwitch adminstate.SwitchRoutingCommand lastSwitch adminstate.SwitchRoutingCommand
lastDisable adminstate.DisableRoutingCommand
lastConfig adminstate.CommitConfigCommand lastConfig adminstate.CommitConfigCommand
onCommit func(adminstate.CommitConfigCommand) onCommit func(adminstate.CommitConfigCommand)
} }
@ -665,6 +666,10 @@ func (state *orderedCommitState) SwitchRouting(context.Context, adminstate.Switc
return adminstate.MutationResult{}, nil 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) { func (state *orderedCommitState) CommitConfig(_ context.Context, command adminstate.CommitConfigCommand) (adminstate.MutationResult, error) {
revision := state.next.Add(1) revision := state.next.Add(1)
if revision == 1 { if revision == 1 {
@ -688,6 +693,11 @@ func (state *recordingAdminState) SwitchRouting(_ context.Context, command admin
return state.mutation, state.err 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) { func (state *recordingAdminState) CommitConfig(_ context.Context, command adminstate.CommitConfigCommand) (adminstate.MutationResult, error) {
state.lastConfig = command state.lastConfig = command
if state.onCommit != nil { if state.onCommit != nil {

View File

@ -25,6 +25,7 @@ const (
defaultPollInterval = time.Second defaultPollInterval = time.Second
autoSwitchActor = "proxy-controller" autoSwitchActor = "proxy-controller"
autoSwitchReason = "consecutive empty provider fetches reached routing threshold" 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 // ConfigurationSource supplies one immutable configuration and its matching
@ -38,6 +39,7 @@ type ConfigurationSource interface {
type StateStore interface { type StateStore interface {
Snapshot(context.Context) (adminstate.Snapshot, error) Snapshot(context.Context) (adminstate.Snapshot, error)
SwitchRouting(context.Context, adminstate.SwitchRoutingCommand) (adminstate.MutationResult, error) SwitchRouting(context.Context, adminstate.SwitchRoutingCommand) (adminstate.MutationResult, error)
DisableRouting(context.Context, adminstate.DisableRoutingCommand) (adminstate.MutationResult, error)
} }
type SnapshotRefreshNotifier interface { type SnapshotRefreshNotifier interface {
@ -51,6 +53,7 @@ type Options struct {
type TickResult struct { type TickResult struct {
Switched int Switched int
Stopped int
} }
// SequentialCoordinator observes Provider fetch outcomes without blocking the // 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 { if coordinator.processed[key] == stats.EmptyGeneration {
continue continue
} }
target, switchable := nextUpstream(route, state.CurrentUpstream, upstreams, configuration) transition := nextTransition(route, state.CurrentUpstream, upstreams, configuration)
if !switchable { 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 coordinator.processed[key] = stats.EmptyGeneration
continue continue
} }
mutation, mutateErr := coordinator.state.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{ 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(), 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 mutateErr != nil {
if errors.Is(mutateErr, adminstate.ErrConflict) { if errors.Is(mutateErr, adminstate.ErrConflict) {
@ -207,12 +232,17 @@ func (coordinator *SequentialCoordinator) Tick(ctx context.Context) (TickResult,
return result, nil return result, nil
} }
func nextUpstream( type sequentialTransition struct {
target string
stop bool
}
func nextTransition(
route config.Routing, route config.Routing,
current string, current string,
states map[string]adminstate.UpstreamState, states map[string]adminstate.UpstreamState,
configuration *config.Config, configuration *config.Config,
) (string, bool) { ) sequentialTransition {
eligible := make([]string, 0, len(route.Upstreams)) eligible := make([]string, 0, len(route.Upstreams))
currentIndex := -1 currentIndex := -1
for _, upstream := range route.Upstreams { for _, upstream := range route.Upstreams {
@ -224,20 +254,25 @@ func nextUpstream(
} }
eligible = append(eligible, upstream) eligible = append(eligible, upstream)
} }
if currentIndex < 0 || len(eligible) < 2 { if currentIndex < 0 {
return "", false return sequentialTransition{}
} }
if currentIndex+1 < len(eligible) { if currentIndex+1 < len(eligible) {
return eligible[currentIndex+1], true return sequentialTransition{target: eligible[currentIndex+1]}
} }
switch route.Strategy.EndBehavior { switch route.Strategy.EndBehavior {
case "loop": case "loop":
return eligible[0], eligible[0] != current if eligible[0] != current {
case "stayLast", "", "stop": return sequentialTransition{target: eligible[0]}
return "", false }
case "stop", "":
return sequentialTransition{stop: true}
case "stayLast":
return sequentialTransition{}
default: default:
return "", false return sequentialTransition{}
} }
return sequentialTransition{}
} }
func upstreamStates(values []adminstate.UpstreamState) map[string]adminstate.UpstreamState { func upstreamStates(values []adminstate.UpstreamState) map[string]adminstate.UpstreamState {
@ -273,8 +308,8 @@ func processedKey(routeName, upstream string) string {
return routeName + "\x00" + upstream return routeName + "\x00" + upstream
} }
func requestID(routeName, upstream string, generation uint64) string { func requestID(action, routeName, upstream string, generation uint64) string {
payload := routeName + "\x00" + upstream + "\x00" + strconv.FormatUint(generation, 10) payload := action + "\x00" + routeName + "\x00" + upstream + "\x00" + strconv.FormatUint(generation, 10)
digest := sha256.Sum256([]byte(payload)) digest := sha256.Sum256([]byte(payload))
return "auto-sequential-" + hex.EncodeToString(digest[:16]) return "auto-sequential-" + hex.EncodeToString(digest[:16])
} }

View File

@ -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) { func TestSequentialCoordinatorConsumesStatsRecorderNotifications(t *testing.T) {
state := newRoutingState("provider-a", map[string]bool{"provider-a": true, "provider-b": true}) state := newRoutingState("provider-a", map[string]bool{"provider-a": true, "provider-b": true})
stats, err := provider.NewStatsRecorder(2) stats, err := provider.NewStatsRecorder(2)
@ -206,6 +234,7 @@ type routingState struct {
mu sync.Mutex mu sync.Mutex
snapshot adminstate.Snapshot snapshot adminstate.Snapshot
switches int switches int
stops int
} }
func newRoutingState(current string, enabled map[string]bool) *routingState { 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 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 { func (state *routingState) current() string {
state.mu.Lock() state.mu.Lock()
defer state.mu.Unlock() defer state.mu.Unlock()
return state.snapshot.Routings[0].CurrentUpstream 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 { type routingStats struct {
mu sync.Mutex mu sync.Mutex
byUpstream map[string]provider.Stats byUpstream map[string]provider.Stats

View File

@ -35,6 +35,7 @@ var identifierPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$`)
type Mutator interface { type Mutator interface {
SetUpstreamEnabled(context.Context, SetUpstreamCommand) (MutationResult, error) SetUpstreamEnabled(context.Context, SetUpstreamCommand) (MutationResult, error)
SwitchRouting(context.Context, SwitchRoutingCommand) (MutationResult, error) SwitchRouting(context.Context, SwitchRoutingCommand) (MutationResult, error)
DisableRouting(context.Context, DisableRoutingCommand) (MutationResult, error)
CommitConfig(context.Context, CommitConfigCommand) (MutationResult, error) CommitConfig(context.Context, CommitConfigCommand) (MutationResult, error)
} }
@ -81,6 +82,18 @@ type SwitchRoutingCommand struct {
Reason string 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 { type CommitConfigCommand struct {
RequestID string RequestID string
Actor Actor Actor Actor
@ -147,6 +160,7 @@ type Action string
const ( const (
ActionSetUpstream Action = "set_upstream_enabled" ActionSetUpstream Action = "set_upstream_enabled"
ActionSwitchRoute Action = "switch_routing" ActionSwitchRoute Action = "switch_routing"
ActionDisableRoute Action = "disable_routing"
ActionCommitConfig Action = "commit_config" ActionCommitConfig Action = "commit_config"
) )
@ -213,6 +227,17 @@ func (command SwitchRoutingCommand) Validate() error {
return nil 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. // Validate checks the complete management snapshot and all of its references.
func (command CommitConfigCommand) Validate() error { func (command CommitConfigCommand) Validate() error {
if err := validateMutationBase(command.RequestID, command.Actor, command.OccurredAt); err != nil || if err := validateMutationBase(command.RequestID, command.Actor, command.OccurredAt); err != nil ||

View File

@ -29,6 +29,9 @@ func Run(t *testing.T, factory Factory) {
t.Run("routing no-op audit", func(t *testing.T) { t.Run("routing no-op audit", func(t *testing.T) {
runRoutingNoOpContract(t, factory(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) { t.Run("audit pagination and fields", func(t *testing.T) {
runAuditContract(t, factory(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) { func runAuditContract(t *testing.T, store adminstate.Store) {
t.Helper() t.Helper()
now := contractNow() now := contractNow()
@ -355,6 +446,12 @@ func runContextContract(t *testing.T, store adminstate.Store) {
}); !errors.Is(err, context.Canceled) { }); !errors.Is(err, context.Canceled) {
t.Fatalf("SwitchRouting(canceled) error = %v", err) 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) { if _, err := store.Snapshot(ctx); !errors.Is(err, context.Canceled) {
t.Fatalf("Snapshot(canceled) error = %v", err) t.Fatalf("Snapshot(canceled) error = %v", err)
} }
@ -411,3 +508,12 @@ func upstreamEnabled(snapshot adminstate.Snapshot, name string) bool {
} }
return false return false
} }
func routingEnabled(snapshot adminstate.Snapshot, name string) bool {
for _, routing := range snapshot.Routings {
if routing.Name == name {
return routing.Enabled
}
}
return false
}

View File

@ -210,6 +210,60 @@ func (store *MemoryStore) SwitchRouting(ctx context.Context, command SwitchRouti
return MutationResult{RequestID: command.RequestID, Changed: true, Revision: nextRevision}, nil 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) { func (store *MemoryStore) Snapshot(ctx context.Context) (Snapshot, error) {
if err := contextError(ctx); err != nil { if err := contextError(ctx); err != nil {
return Snapshot{}, err return Snapshot{}, err

View File

@ -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) { func TestValidateCommitConfigCommandAndReferences(t *testing.T) {
t.Parallel() t.Parallel()
valid := validCommitConfigCommand() valid := validCommitConfigCommand()