feat: define admin state transaction contracts

This commit is contained in:
youfak 2026-07-29 18:21:21 +08:00
parent 08800cb7a5
commit 801712ff24
2 changed files with 587 additions and 0 deletions

View File

@ -0,0 +1,388 @@
package adminstate
import (
"context"
"encoding/hex"
"encoding/json"
"errors"
"net/netip"
"regexp"
"strings"
"time"
"unicode"
"unicode/utf8"
)
const (
MaxIdentifierBytes = 128
MaxActorIDBytes = 256
MaxReasonBytes = 512
MaxSourceBytes = 512
MaxDefinitions = 10_000
MaxPageSize = 1_000
SHA256HexBytes = 64
)
var (
ErrInvalidCommand = errors.New("invalid admin state command")
ErrNotFound = errors.New("admin state resource not found")
ErrConflict = errors.New("admin state conflict")
ErrUnavailable = errors.New("admin state unavailable")
)
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)
CommitConfig(context.Context, CommitConfigCommand) (MutationResult, error)
}
type SnapshotReader interface {
Snapshot(context.Context) (Snapshot, error)
}
type AuditReader interface {
ReadAudit(context.Context, AuditQuery) ([]AuditRecord, error)
}
type Outbox interface {
Claim(context.Context, ClaimCommand) ([]Event, error)
Acknowledge(context.Context, AcknowledgeCommand) error
}
type Store interface {
Mutator
SnapshotReader
AuditReader
Outbox
}
type Actor struct {
ID string
SourceIP string
}
type SetUpstreamCommand struct {
RequestID string
Actor Actor
OccurredAt time.Time
Name string
Enabled bool
}
type SwitchRoutingCommand struct {
RequestID string
Actor Actor
OccurredAt time.Time
Name string
ExpectedCurrent string
Target string
Reason string
}
type CommitConfigCommand struct {
RequestID string
Actor Actor
OccurredAt time.Time
ConfigVersion string
Checksum string
Source string
Upstreams []UpstreamDefinition
Routings []RoutingDefinition
}
type UpstreamDefinition struct {
Name string
Enabled bool
}
type RoutingDefinition struct {
Name string
Enabled bool
Upstreams []string
CurrentUpstream string
}
type MutationResult struct {
RequestID string
Changed bool
Revision uint64
Message string
}
type ConfigRevision struct {
Revision uint64
ConfigVersion string
Checksum string
Source string
CreatedAt time.Time
}
type UpstreamState struct {
Name string
Enabled bool
Revision uint64
UpdatedAt time.Time
}
type RoutingState struct {
Name string
Enabled bool
Upstreams []string
CurrentUpstream string
Revision uint64
UpdatedAt time.Time
}
type Snapshot struct {
Revision uint64
Config *ConfigRevision
Upstreams []UpstreamState
Routings []RoutingState
}
type Action string
const (
ActionSetUpstream Action = "set_upstream_enabled"
ActionSwitchRoute Action = "switch_routing"
ActionCommitConfig Action = "commit_config"
)
type AuditRecord struct {
ID uint64
RequestID string
Actor Actor
Action Action
ResourceType string
ResourceName string
Changed bool
Revision uint64
Reason string
OccurredAt time.Time
}
type AuditQuery struct {
AfterID uint64
Limit int
}
type Event struct {
ID uint64
Revision uint64
Type string
AggregateType string
AggregateID string
Payload json.RawMessage
OccurredAt time.Time
ClaimedBy string
ClaimUntil *time.Time
PublishedAt *time.Time
}
type ClaimCommand struct {
ConsumerID string
Now time.Time
Limit int
Lease time.Duration
}
type AcknowledgeCommand struct {
ConsumerID string
Now time.Time
EventIDs []uint64
}
func validateSetUpstreamCommand(command SetUpstreamCommand) error {
if err := validateMutationBase(command.RequestID, command.Actor, command.OccurredAt); err != nil ||
!validIdentifier(command.Name) {
return ErrInvalidCommand
}
return nil
}
func validateSwitchRoutingCommand(command SwitchRoutingCommand) error {
if err := validateMutationBase(command.RequestID, command.Actor, command.OccurredAt); err != nil ||
!validIdentifier(command.Name) || !validIdentifier(command.ExpectedCurrent) ||
!validIdentifier(command.Target) || !validOptionalText(command.Reason, MaxReasonBytes) {
return ErrInvalidCommand
}
return nil
}
func validateCommitConfigCommand(command CommitConfigCommand) error {
if err := validateMutationBase(command.RequestID, command.Actor, command.OccurredAt); err != nil ||
!validRequiredText(command.ConfigVersion, MaxIdentifierBytes) ||
!validSHA256(command.Checksum) || !validRequiredText(command.Source, MaxSourceBytes) ||
len(command.Upstreams) > MaxDefinitions || len(command.Routings) > MaxDefinitions {
return ErrInvalidCommand
}
upstreamNames := make(map[string]struct{}, len(command.Upstreams))
for _, upstream := range command.Upstreams {
if !validIdentifier(upstream.Name) {
return ErrInvalidCommand
}
if _, duplicate := upstreamNames[upstream.Name]; duplicate {
return ErrInvalidCommand
}
upstreamNames[upstream.Name] = struct{}{}
}
routingNames := make(map[string]struct{}, len(command.Routings))
for _, routing := range command.Routings {
if !validIdentifier(routing.Name) || len(routing.Upstreams) == 0 ||
len(routing.Upstreams) > MaxDefinitions || !validIdentifier(routing.CurrentUpstream) {
return ErrInvalidCommand
}
if _, duplicate := routingNames[routing.Name]; duplicate {
return ErrInvalidCommand
}
routingNames[routing.Name] = struct{}{}
candidates := make(map[string]struct{}, len(routing.Upstreams))
for _, upstream := range routing.Upstreams {
if !validIdentifier(upstream) {
return ErrInvalidCommand
}
if _, exists := upstreamNames[upstream]; !exists {
return ErrInvalidCommand
}
if _, duplicate := candidates[upstream]; duplicate {
return ErrInvalidCommand
}
candidates[upstream] = struct{}{}
}
if _, exists := candidates[routing.CurrentUpstream]; !exists {
return ErrInvalidCommand
}
}
return nil
}
func validateClaimCommand(command ClaimCommand) error {
if !validIdentifier(command.ConsumerID) || command.Now.IsZero() || command.Limit <= 0 ||
command.Limit > MaxPageSize || command.Lease <= 0 {
return ErrInvalidCommand
}
return nil
}
func validateAcknowledgeCommand(command AcknowledgeCommand) error {
if !validIdentifier(command.ConsumerID) || command.Now.IsZero() || len(command.EventIDs) == 0 ||
len(command.EventIDs) > MaxPageSize {
return ErrInvalidCommand
}
seen := make(map[uint64]struct{}, len(command.EventIDs))
for _, eventID := range command.EventIDs {
if eventID == 0 {
return ErrInvalidCommand
}
if _, duplicate := seen[eventID]; duplicate {
return ErrInvalidCommand
}
seen[eventID] = struct{}{}
}
return nil
}
func validateAuditQuery(query AuditQuery) error {
if query.Limit <= 0 || query.Limit > MaxPageSize {
return ErrInvalidCommand
}
return nil
}
func validateMutationBase(requestID string, actor Actor, occurredAt time.Time) error {
if !validRequiredText(requestID, MaxIdentifierBytes) || !validRequiredText(actor.ID, MaxActorIDBytes) ||
occurredAt.IsZero() {
return ErrInvalidCommand
}
if actor.SourceIP != "" {
address, err := netip.ParseAddr(actor.SourceIP)
if err != nil || !address.IsValid() {
return ErrInvalidCommand
}
}
return nil
}
func validIdentifier(value string) bool {
return len(value) <= MaxIdentifierBytes && identifierPattern.MatchString(value)
}
func validSHA256(value string) bool {
if len(value) != SHA256HexBytes {
return false
}
decoded, err := hex.DecodeString(value)
return err == nil && len(decoded) == SHA256HexBytes/2
}
func validRequiredText(value string, maximum int) bool {
return value != "" && validOptionalText(value, maximum)
}
func validOptionalText(value string, maximum int) bool {
if len(value) > maximum || !utf8.ValidString(value) || strings.TrimSpace(value) != value {
return false
}
for _, character := range value {
if unicode.IsControl(character) {
return false
}
}
return true
}
func cloneCommitConfigCommand(command CommitConfigCommand) CommitConfigCommand {
cloned := command
cloned.Upstreams = append([]UpstreamDefinition(nil), command.Upstreams...)
cloned.Routings = make([]RoutingDefinition, len(command.Routings))
for index, routing := range command.Routings {
cloned.Routings[index] = routing
cloned.Routings[index].Upstreams = append([]string(nil), routing.Upstreams...)
}
return cloned
}
func cloneSnapshot(snapshot Snapshot) Snapshot {
cloned := snapshot
if snapshot.Config != nil {
config := *snapshot.Config
cloned.Config = &config
}
cloned.Upstreams = append([]UpstreamState(nil), snapshot.Upstreams...)
cloned.Routings = make([]RoutingState, len(snapshot.Routings))
for index, routing := range snapshot.Routings {
cloned.Routings[index] = routing
cloned.Routings[index].Upstreams = append([]string(nil), routing.Upstreams...)
}
return cloned
}
func cloneEvent(event Event) Event {
cloned := event
cloned.Payload = append(json.RawMessage(nil), event.Payload...)
if event.ClaimUntil != nil {
value := *event.ClaimUntil
cloned.ClaimUntil = &value
}
if event.PublishedAt != nil {
value := *event.PublishedAt
cloned.PublishedAt = &value
}
return cloned
}
func cloneEvents(events []Event) []Event {
cloned := make([]Event, len(events))
for index, event := range events {
cloned[index] = cloneEvent(event)
}
return cloned
}
func cloneAuditRecords(records []AuditRecord) []AuditRecord {
return append([]AuditRecord(nil), records...)
}

View File

@ -0,0 +1,199 @@
package adminstate
import (
"encoding/json"
"errors"
"strings"
"testing"
"time"
)
func TestValidateSetUpstreamCommand(t *testing.T) {
t.Parallel()
now := time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC)
valid := SetUpstreamCommand{
RequestID: "req-upstream", Actor: Actor{ID: "admin-a", SourceIP: "192.0.2.10"},
OccurredAt: now, Name: "provider-a", Enabled: true,
}
if err := validateSetUpstreamCommand(valid); err != nil {
t.Fatalf("validateSetUpstreamCommand(valid): %v", err)
}
tests := []struct {
name string
mutate func(*SetUpstreamCommand)
}{
{name: "missing request", mutate: func(command *SetUpstreamCommand) { command.RequestID = "" }},
{name: "missing actor", mutate: func(command *SetUpstreamCommand) { command.Actor.ID = "" }},
{name: "invalid source", mutate: func(command *SetUpstreamCommand) { command.Actor.SourceIP = "not-an-ip" }},
{name: "zero time", mutate: func(command *SetUpstreamCommand) { command.OccurredAt = time.Time{} }},
{name: "invalid name", mutate: func(command *SetUpstreamCommand) { command.Name = "provider/a" }},
}
for _, test := range tests {
test := test
t.Run(test.name, func(t *testing.T) {
t.Parallel()
command := valid
test.mutate(&command)
if err := validateSetUpstreamCommand(command); !errors.Is(err, ErrInvalidCommand) {
t.Fatalf("validateSetUpstreamCommand() error = %v, want ErrInvalidCommand", err)
}
})
}
}
func TestValidateSwitchRoutingCommand(t *testing.T) {
t.Parallel()
now := time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC)
valid := SwitchRoutingCommand{
RequestID: "req-switch", Actor: Actor{ID: "admin-a"}, OccurredAt: now,
Name: "checkout", ExpectedCurrent: "provider-a", Target: "provider-b", Reason: "capacity",
}
if err := validateSwitchRoutingCommand(valid); err != nil {
t.Fatalf("validateSwitchRoutingCommand(valid): %v", err)
}
for _, test := range []struct {
name string
mutate func(*SwitchRoutingCommand)
}{
{name: "missing routing", mutate: func(command *SwitchRoutingCommand) { command.Name = "" }},
{name: "missing expected", mutate: func(command *SwitchRoutingCommand) { command.ExpectedCurrent = "" }},
{name: "missing target", mutate: func(command *SwitchRoutingCommand) { command.Target = "" }},
{name: "reason too long", mutate: func(command *SwitchRoutingCommand) { 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 := validateSwitchRoutingCommand(command); !errors.Is(err, ErrInvalidCommand) {
t.Fatalf("validateSwitchRoutingCommand() error = %v, want ErrInvalidCommand", err)
}
})
}
}
func TestValidateCommitConfigCommandAndReferences(t *testing.T) {
t.Parallel()
valid := validCommitConfigCommand()
if err := validateCommitConfigCommand(valid); err != nil {
t.Fatalf("validateCommitConfigCommand(valid): %v", err)
}
for _, test := range []struct {
name string
mutate func(*CommitConfigCommand)
}{
{name: "invalid checksum", mutate: func(command *CommitConfigCommand) { command.Checksum = "sha256:bad" }},
{name: "duplicate upstream", mutate: func(command *CommitConfigCommand) {
command.Upstreams = append(command.Upstreams, command.Upstreams[0])
}},
{name: "duplicate routing", mutate: func(command *CommitConfigCommand) { command.Routings = append(command.Routings, command.Routings[0]) }},
{name: "duplicate candidate", mutate: func(command *CommitConfigCommand) {
command.Routings[0].Upstreams = append(command.Routings[0].Upstreams, "provider-a")
}},
{name: "unknown candidate", mutate: func(command *CommitConfigCommand) { command.Routings[0].Upstreams[0] = "provider-missing" }},
{name: "current not candidate", mutate: func(command *CommitConfigCommand) { command.Routings[0].CurrentUpstream = "provider-c" }},
{name: "empty candidate list", mutate: func(command *CommitConfigCommand) { command.Routings[0].Upstreams = nil }},
} {
test := test
t.Run(test.name, func(t *testing.T) {
t.Parallel()
command := cloneCommitConfigCommand(valid)
test.mutate(&command)
if err := validateCommitConfigCommand(command); !errors.Is(err, ErrInvalidCommand) {
t.Fatalf("validateCommitConfigCommand() error = %v, want ErrInvalidCommand", err)
}
})
}
}
func TestValidateOutboxCommands(t *testing.T) {
t.Parallel()
now := time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC)
if err := validateClaimCommand(ClaimCommand{
ConsumerID: "publisher-a", Now: now, Limit: 100, Lease: time.Minute,
}); err != nil {
t.Fatalf("validateClaimCommand(valid): %v", err)
}
if err := validateAcknowledgeCommand(AcknowledgeCommand{
ConsumerID: "publisher-a", Now: now, EventIDs: []uint64{1, 2},
}); err != nil {
t.Fatalf("validateAcknowledgeCommand(valid): %v", err)
}
invalidClaims := []ClaimCommand{
{Now: now, Limit: 1, Lease: time.Second},
{ConsumerID: "publisher-a", Limit: 1, Lease: time.Second},
{ConsumerID: "publisher-a", Now: now, Lease: time.Second},
{ConsumerID: "publisher-a", Now: now, Limit: MaxPageSize + 1, Lease: time.Second},
{ConsumerID: "publisher-a", Now: now, Limit: 1},
}
for _, command := range invalidClaims {
if err := validateClaimCommand(command); !errors.Is(err, ErrInvalidCommand) {
t.Fatalf("validateClaimCommand(%+v) error = %v", command, err)
}
}
invalidAcks := []AcknowledgeCommand{
{Now: now, EventIDs: []uint64{1}},
{ConsumerID: "publisher-a", EventIDs: []uint64{1}},
{ConsumerID: "publisher-a", Now: now},
{ConsumerID: "publisher-a", Now: now, EventIDs: []uint64{0}},
{ConsumerID: "publisher-a", Now: now, EventIDs: []uint64{1, 1}},
}
for _, command := range invalidAcks {
if err := validateAcknowledgeCommand(command); !errors.Is(err, ErrInvalidCommand) {
t.Fatalf("validateAcknowledgeCommand(%+v) error = %v", command, err)
}
}
}
func TestCloneValuesDoNotShareMutableState(t *testing.T) {
t.Parallel()
command := validCommitConfigCommand()
clonedCommand := cloneCommitConfigCommand(command)
clonedCommand.Upstreams[0].Name = "changed"
clonedCommand.Routings[0].Upstreams[0] = "changed"
if command.Upstreams[0].Name != "provider-a" || command.Routings[0].Upstreams[0] != "provider-a" {
t.Fatalf("cloneCommitConfigCommand shared input storage: %+v", command)
}
snapshot := Snapshot{
Revision: 3,
Config: &ConfigRevision{Revision: 3, ConfigVersion: "cfg-3"},
Upstreams: []UpstreamState{{Name: "provider-a", Enabled: true}},
Routings: []RoutingState{{Name: "checkout", Upstreams: []string{"provider-a"}, CurrentUpstream: "provider-a"}},
}
clonedSnapshot := cloneSnapshot(snapshot)
clonedSnapshot.Config.ConfigVersion = "changed"
clonedSnapshot.Upstreams[0].Name = "changed"
clonedSnapshot.Routings[0].Upstreams[0] = "changed"
if snapshot.Config.ConfigVersion != "cfg-3" || snapshot.Upstreams[0].Name != "provider-a" ||
snapshot.Routings[0].Upstreams[0] != "provider-a" {
t.Fatalf("cloneSnapshot shared input storage: %+v", snapshot)
}
event := Event{Payload: json.RawMessage(`{"enabled":true}`)}
clonedEvent := cloneEvent(event)
clonedEvent.Payload[2] = 'X'
if string(event.Payload) != `{"enabled":true}` {
t.Fatalf("cloneEvent shared payload storage: %s", event.Payload)
}
}
func validCommitConfigCommand() CommitConfigCommand {
return CommitConfigCommand{
RequestID: "req-config", Actor: Actor{ID: "admin-a", SourceIP: "192.0.2.10"},
OccurredAt: time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC),
ConfigVersion: "cfg-1", Checksum: strings.Repeat("a", SHA256HexBytes), Source: "configs/proxy-pool.yaml",
Upstreams: []UpstreamDefinition{
{Name: "provider-a", Enabled: true},
{Name: "provider-b", Enabled: true},
},
Routings: []RoutingDefinition{{
Name: "checkout", Enabled: true, Upstreams: []string{"provider-a", "provider-b"}, CurrentUpstream: "provider-a",
}},
}
}