From 2341296271d39fd5ace107eaf7a81795594b7682 Mon Sep 17 00:00:00 2001 From: youfak Date: Wed, 29 Jul 2026 21:50:21 +0800 Subject: [PATCH] test: strengthen admin state contract --- docs/adr/006-postgresql-admin-state.md | 3 + .../adminstate/contracttest/contract.go | 117 ++++++++++++++++++ progress.md | 3 + 3 files changed, 123 insertions(+) diff --git a/docs/adr/006-postgresql-admin-state.md b/docs/adr/006-postgresql-admin-state.md index ee93193..8afca90 100644 --- a/docs/adr/006-postgresql-admin-state.md +++ b/docs/adr/006-postgresql-admin-state.md @@ -127,6 +127,9 @@ MemoryStore 与 PostgreSQL Adapter 运行相同契约，至少覆盖： - Routing CAS、目标校验和 100 个并发请求最多一个成功。 - 任一审计/Outbox 写故障导致状态完全回滚。 - Outbox 有界 claim、租约到期重试、错误 consumer ACK 拒绝和顺序稳定。 +- 审计按 ID 稳定分页并保留 Actor、资源、动作、修订和 UTC 时间；Routing no-op + 写审计但不写 Outbox。 +- 批量 ACK 先完整校验所有事件再提交，任一未知或冲突 ID 不得部分发布。 - 上下文取消、错误脱敏和 Snapshot 防止调用方修改内部状态。 - 真实 PostgreSQL 重复迁移、事务回滚和禁止数据表边界。 diff --git a/internal/domain/adminstate/contracttest/contract.go b/internal/domain/adminstate/contracttest/contract.go index 4c401f4..8b90ea0 100644 --- a/internal/domain/adminstate/contracttest/contract.go +++ b/internal/domain/adminstate/contracttest/contract.go @@ -26,9 +26,18 @@ func Run(t *testing.T, factory Factory) { t.Run("routing compare and swap", func(t *testing.T) { runRoutingContract(t, factory(t)) }) + t.Run("routing no-op audit", func(t *testing.T) { + runRoutingNoOpContract(t, factory(t)) + }) + t.Run("audit pagination and fields", func(t *testing.T) { + runAuditContract(t, factory(t)) + }) t.Run("outbox lease and acknowledgement", func(t *testing.T) { runOutboxContract(t, factory(t)) }) + t.Run("outbox acknowledgement is atomic", func(t *testing.T) { + runAtomicAcknowledgeContract(t, factory(t)) + }) t.Run("context cancellation", func(t *testing.T) { runContextContract(t, factory(t)) }) @@ -185,6 +194,67 @@ func runRoutingContract(t *testing.T, store adminstate.Store) { } } +func runRoutingNoOpContract(t *testing.T, store adminstate.Store) { + t.Helper() + now := contractNow() + commit(t, store, configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes))) + + result, err := store.SwitchRouting(context.Background(), adminstate.SwitchRoutingCommand{ + RequestID: "req-switch-noop", Actor: contractActor(), OccurredAt: now.Add(time.Second), + Name: "checkout", ExpectedCurrent: "provider-a", Target: "provider-a", Reason: "already selected", + }) + if err != nil || result.Changed || result.Revision != 1 { + t.Fatalf("SwitchRouting(no-op) = %+v, %v", result, err) + } + audits, err := store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 10}) + if err != nil || len(audits) != 2 || audits[1].Action != adminstate.ActionSwitchRoute || + audits[1].Changed || audits[1].Revision != 1 || audits[1].Reason != "already selected" { + t.Fatalf("ReadAudit(no-op) = %+v, %v", audits, err) + } + 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) != 1 || events[0].Type != "config.committed" { + t.Fatalf("Claim(after no-op) = %+v, %v", events, err) + } +} + +func runAuditContract(t *testing.T, store adminstate.Store) { + t.Helper() + now := contractNow() + commit(t, store, configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes))) + _, err := store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{ + RequestID: "req-disable", Actor: adminstate.Actor{ID: "admin-b", SourceIP: "::ffff:192.0.2.11"}, + OccurredAt: now.Add(time.Second), Name: "provider-a", Enabled: false, + }) + if err != nil { + t.Fatalf("SetUpstreamEnabled(): %v", err) + } + + first, err := store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 1}) + if err != nil || len(first) != 1 { + t.Fatalf("ReadAudit(first page) = %+v, %v", first, err) + } + if first[0].ID == 0 || first[0].RequestID != "req-config" || first[0].Actor != contractActor() || + first[0].Action != adminstate.ActionCommitConfig || first[0].ResourceType != "config" || + first[0].ResourceName != "cfg-1" || !first[0].Changed || first[0].Revision != 1 || + !first[0].OccurredAt.Equal(now) { + t.Fatalf("first audit record = %+v", first[0]) + } + second, err := store.ReadAudit(context.Background(), adminstate.AuditQuery{AfterID: first[0].ID, Limit: 1}) + if err != nil || len(second) != 1 || second[0].ID <= first[0].ID || second[0].RequestID != "req-disable" || + second[0].Actor.ID != "admin-b" || second[0].Actor.SourceIP != "192.0.2.11" || + second[0].Action != adminstate.ActionSetUpstream || second[0].ResourceType != "upstream" || + second[0].ResourceName != "provider-a" || !second[0].Changed || second[0].Revision != 2 || + !second[0].OccurredAt.Equal(now.Add(time.Second)) { + t.Fatalf("ReadAudit(second page) = %+v, %v", second, err) + } + empty, err := store.ReadAudit(context.Background(), adminstate.AuditQuery{AfterID: second[0].ID, Limit: 1}) + if err != nil || len(empty) != 0 { + t.Fatalf("ReadAudit(after end) = %+v, %v", empty, err) + } +} + func runOutboxContract(t *testing.T, store adminstate.Store) { t.Helper() now := contractNow() @@ -235,6 +305,36 @@ func runOutboxContract(t *testing.T, store adminstate.Store) { } } +func runAtomicAcknowledgeContract(t *testing.T, store adminstate.Store) { + t.Helper() + now := contractNow() + commit(t, store, configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes))) + if _, err := store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{ + RequestID: "req-disable", Actor: contractActor(), OccurredAt: now.Add(time.Second), + Name: "provider-a", Enabled: false, + }); err != nil { + t.Fatalf("SetUpstreamEnabled(): %v", err) + } + 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 { + t.Fatalf("Claim() = %+v, %v", events, err) + } + if err := store.Acknowledge(context.Background(), adminstate.AcknowledgeCommand{ + ConsumerID: "publisher-a", Now: now.Add(3 * time.Second), EventIDs: []uint64{events[0].ID, events[1].ID + 1000}, + }); !errors.Is(err, adminstate.ErrNotFound) { + t.Fatalf("Acknowledge(partially invalid) error = %v", err) + } + for _, event := range events { + if err := store.Acknowledge(context.Background(), adminstate.AcknowledgeCommand{ + ConsumerID: "publisher-a", Now: now.Add(4 * time.Second), EventIDs: []uint64{event.ID}, + }); err != nil { + t.Fatalf("Acknowledge(%d) after failed batch: %v", event.ID, err) + } + } +} + func runContextContract(t *testing.T, store adminstate.Store) { t.Helper() ctx, cancel := context.WithCancel(context.Background()) @@ -243,6 +343,18 @@ func runContextContract(t *testing.T, store adminstate.Store) { if _, err := store.CommitConfig(ctx, command); !errors.Is(err, context.Canceled) { t.Fatalf("CommitConfig(canceled) error = %v", err) } + if _, err := store.SetUpstreamEnabled(ctx, adminstate.SetUpstreamCommand{ + RequestID: "req-upstream", Actor: contractActor(), OccurredAt: contractNow(), + Name: "provider-a", Enabled: false, + }); !errors.Is(err, context.Canceled) { + t.Fatalf("SetUpstreamEnabled(canceled) error = %v", err) + } + if _, err := store.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{ + RequestID: "req-switch", Actor: contractActor(), OccurredAt: contractNow(), + Name: "checkout", ExpectedCurrent: "provider-a", Target: "provider-b", + }); !errors.Is(err, context.Canceled) { + t.Fatalf("SwitchRouting(canceled) error = %v", err) + } if _, err := store.Snapshot(ctx); !errors.Is(err, context.Canceled) { t.Fatalf("Snapshot(canceled) error = %v", err) } @@ -254,6 +366,11 @@ func runContextContract(t *testing.T, store adminstate.Store) { }); !errors.Is(err, context.Canceled) { t.Fatalf("Claim(canceled) error = %v", err) } + if err := store.Acknowledge(ctx, adminstate.AcknowledgeCommand{ + ConsumerID: "publisher-a", Now: contractNow(), EventIDs: []uint64{1}, + }); !errors.Is(err, context.Canceled) { + t.Fatalf("Acknowledge(canceled) error = %v", err) + } } func configCommand(requestID, version, checksum string) adminstate.CommitConfigCommand { diff --git a/progress.md b/progress.md index 4f0221b..9b62093 100644 --- a/progress.md +++ b/progress.md @@ -2,6 +2,9 @@ ## 2026-07-29 +- 扩展 `adminstate/contracttest` 公用契约：审计完整字段和 AfterID 分页、Routing + no-op 审计且不写 Outbox、批量 ACK 零部分提交，以及六个 Store 方法的 Context + 取消；MemoryStore 全部通过，后续 PostgreSQL Adapter 必须运行同一套契约。 - `adminstate` 六类命令/查询已封装公用 `Validate()`，MemoryStore 改为统一复用； 后续 pgx Adapter 不再重复实现名称、引用、分页和 Outbox 租约输入校验。 - 一次命令装配审计误用了不存在的 `controller/distribution/service.go`；实际领域