feat: add transactional admin state reference store
This commit is contained in:
youfak
parent
801712ff24
commit
b53b9f1adc
15
internal/domain/adminstate/contract_external_test.go
Normal file
15
internal/domain/adminstate/contract_external_test.go
Normal file
@ -0,0 +1,15 @@
|
||||
package adminstate_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"proxy-pool/internal/domain/adminstate"
|
||||
"proxy-pool/internal/domain/adminstate/contracttest"
|
||||
)
|
||||
|
||||
func TestMemoryStoreContract(t *testing.T) {
|
||||
contracttest.Run(t, func(t *testing.T) adminstate.Store {
|
||||
t.Helper()
|
||||
return adminstate.NewMemoryStore()
|
||||
})
|
||||
}
|
||||
295
internal/domain/adminstate/contracttest/contract.go
Normal file
295
internal/domain/adminstate/contracttest/contract.go
Normal file
@ -0,0 +1,295 @@
|
||||
package contracttest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"proxy-pool/internal/domain/adminstate"
|
||||
)
|
||||
|
||||
type Factory func(*testing.T) adminstate.Store
|
||||
|
||||
func Run(t *testing.T, factory Factory) {
|
||||
t.Helper()
|
||||
t.Run("config transaction and immutability", func(t *testing.T) {
|
||||
runConfigContract(t, factory(t))
|
||||
})
|
||||
t.Run("upstream idempotency", func(t *testing.T) {
|
||||
runUpstreamContract(t, factory(t))
|
||||
})
|
||||
t.Run("routing compare and swap", func(t *testing.T) {
|
||||
runRoutingContract(t, factory(t))
|
||||
})
|
||||
t.Run("outbox lease and acknowledgement", func(t *testing.T) {
|
||||
runOutboxContract(t, factory(t))
|
||||
})
|
||||
t.Run("context cancellation", func(t *testing.T) {
|
||||
runContextContract(t, factory(t))
|
||||
})
|
||||
}
|
||||
|
||||
func runConfigContract(t *testing.T, store adminstate.Store) {
|
||||
t.Helper()
|
||||
command := configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes))
|
||||
result, err := store.CommitConfig(context.Background(), command)
|
||||
if err != nil || !result.Changed || result.Revision != 1 || result.RequestID != command.RequestID {
|
||||
t.Fatalf("CommitConfig(first) = %+v, %v", result, err)
|
||||
}
|
||||
|
||||
snapshot, err := store.Snapshot(context.Background())
|
||||
if err != nil || snapshot.Revision != 1 || snapshot.Config == nil ||
|
||||
snapshot.Config.ConfigVersion != "cfg-1" || len(snapshot.Upstreams) != 2 || len(snapshot.Routings) != 1 {
|
||||
t.Fatalf("Snapshot() = %+v, %v", snapshot, err)
|
||||
}
|
||||
snapshot.Config.ConfigVersion = "mutated"
|
||||
snapshot.Upstreams[0].Name = "mutated"
|
||||
snapshot.Routings[0].Upstreams[0] = "mutated"
|
||||
again, err := store.Snapshot(context.Background())
|
||||
if err != nil || again.Config.ConfigVersion != "cfg-1" ||
|
||||
again.Upstreams[0].Name == "mutated" || again.Routings[0].Upstreams[0] == "mutated" {
|
||||
t.Fatalf("Snapshot() leaked mutable state: %+v, %v", again, err)
|
||||
}
|
||||
|
||||
command.RequestID = "req-config-replay"
|
||||
command.OccurredAt = command.OccurredAt.Add(time.Second)
|
||||
replayed, err := store.CommitConfig(context.Background(), command)
|
||||
if err != nil || replayed.Changed || replayed.Revision != 1 {
|
||||
t.Fatalf("CommitConfig(replay) = %+v, %v", replayed, err)
|
||||
}
|
||||
audits, err := store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 10})
|
||||
if err != nil || len(audits) != 2 || !audits[0].Changed || audits[1].Changed ||
|
||||
audits[0].Revision != 1 || audits[1].Revision != 1 {
|
||||
t.Fatalf("ReadAudit() = %+v, %v", audits, err)
|
||||
}
|
||||
|
||||
conflict := command
|
||||
conflict.RequestID = "req-config-conflict"
|
||||
conflict.Checksum = strings.Repeat("b", adminstate.SHA256HexBytes)
|
||||
if _, err := store.CommitConfig(context.Background(), conflict); !errors.Is(err, adminstate.ErrConflict) {
|
||||
t.Fatalf("CommitConfig(conflict) error = %v", err)
|
||||
}
|
||||
audits, err = store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 10})
|
||||
if err != nil || len(audits) != 2 {
|
||||
t.Fatalf("ReadAudit(after conflict) = %+v, %v", audits, err)
|
||||
}
|
||||
|
||||
invalid := command
|
||||
invalid.RequestID = "req-config-invalid"
|
||||
invalid.Routings[0].CurrentUpstream = "missing"
|
||||
if _, err := store.CommitConfig(context.Background(), invalid); !errors.Is(err, adminstate.ErrInvalidCommand) {
|
||||
t.Fatalf("CommitConfig(invalid) error = %v", err)
|
||||
}
|
||||
final, err := store.Snapshot(context.Background())
|
||||
if err != nil || final.Revision != 1 || final.Config.ConfigVersion != "cfg-1" {
|
||||
t.Fatalf("Snapshot(after invalid) = %+v, %v", final, err)
|
||||
}
|
||||
}
|
||||
|
||||
func runUpstreamContract(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.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{
|
||||
RequestID: "req-disable", Actor: contractActor(), OccurredAt: now.Add(time.Second),
|
||||
Name: "provider-a", Enabled: false,
|
||||
})
|
||||
if err != nil || !result.Changed || result.Revision != 2 {
|
||||
t.Fatalf("SetUpstreamEnabled(disable) = %+v, %v", result, err)
|
||||
}
|
||||
replayed, err := store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{
|
||||
RequestID: "req-disable-replay", Actor: contractActor(), OccurredAt: now.Add(2 * time.Second),
|
||||
Name: "provider-a", Enabled: false,
|
||||
})
|
||||
if err != nil || replayed.Changed || replayed.Revision != 2 {
|
||||
t.Fatalf("SetUpstreamEnabled(replay) = %+v, %v", replayed, err)
|
||||
}
|
||||
if _, err := store.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{
|
||||
RequestID: "req-missing", Actor: contractActor(), OccurredAt: now.Add(3 * time.Second),
|
||||
Name: "missing", Enabled: true,
|
||||
}); !errors.Is(err, adminstate.ErrNotFound) {
|
||||
t.Fatalf("SetUpstreamEnabled(missing) error = %v", err)
|
||||
}
|
||||
|
||||
snapshot, err := store.Snapshot(context.Background())
|
||||
if err != nil || snapshot.Revision != 2 || upstreamEnabled(snapshot, "provider-a") {
|
||||
t.Fatalf("Snapshot() = %+v, %v", snapshot, err)
|
||||
}
|
||||
audits, err := store.ReadAudit(context.Background(), adminstate.AuditQuery{Limit: 10})
|
||||
if err != nil || len(audits) != 3 || !audits[1].Changed || audits[2].Changed {
|
||||
t.Fatalf("ReadAudit() = %+v, %v", audits, err)
|
||||
}
|
||||
events, err := store.Claim(context.Background(), adminstate.ClaimCommand{
|
||||
ConsumerID: "publisher-a", Now: now.Add(4 * time.Second), Limit: 10, Lease: time.Minute,
|
||||
})
|
||||
if err != nil || len(events) != 2 || events[0].Revision != 1 || events[1].Revision != 2 {
|
||||
t.Fatalf("Claim() = %+v, %v", events, err)
|
||||
}
|
||||
}
|
||||
|
||||
func runRoutingContract(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.SwitchRouting(context.Background(), adminstate.SwitchRoutingCommand{
|
||||
RequestID: "req-bad-target", Actor: contractActor(), OccurredAt: now.Add(time.Second),
|
||||
Name: "checkout", ExpectedCurrent: "provider-a", Target: "provider-c",
|
||||
}); !errors.Is(err, adminstate.ErrInvalidCommand) {
|
||||
t.Fatalf("SwitchRouting(bad target) error = %v", err)
|
||||
}
|
||||
|
||||
start := make(chan struct{})
|
||||
var changed atomic.Int64
|
||||
var conflicts 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.SwitchRouting(context.Background(), adminstate.SwitchRoutingCommand{
|
||||
RequestID: fmt.Sprintf("req-switch-%03d", index), Actor: contractActor(),
|
||||
OccurredAt: now.Add(2 * time.Second), Name: "checkout",
|
||||
ExpectedCurrent: "provider-a", Target: "provider-b", Reason: "capacity",
|
||||
})
|
||||
switch {
|
||||
case err == nil && result.Changed:
|
||||
changed.Add(1)
|
||||
case errors.Is(err, adminstate.ErrConflict):
|
||||
conflicts.Add(1)
|
||||
default:
|
||||
unexpectedMu.Lock()
|
||||
unexpected = append(unexpected, err)
|
||||
unexpectedMu.Unlock()
|
||||
}
|
||||
}(index)
|
||||
}
|
||||
close(start)
|
||||
workers.Wait()
|
||||
if changed.Load() != 1 || conflicts.Load() != 99 || len(unexpected) != 0 {
|
||||
t.Fatalf("concurrent switch changed=%d conflicts=%d unexpected=%v", changed.Load(), conflicts.Load(), unexpected)
|
||||
}
|
||||
snapshot, err := store.Snapshot(context.Background())
|
||||
if err != nil || snapshot.Revision != 2 || snapshot.Routings[0].CurrentUpstream != "provider-b" {
|
||||
t.Fatalf("Snapshot() = %+v, %v", snapshot, err)
|
||||
}
|
||||
}
|
||||
|
||||
func runOutboxContract(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.SetUpstreamEnabled(context.Background(), adminstate.SetUpstreamCommand{
|
||||
RequestID: "req-disable", Actor: contractActor(), OccurredAt: now.Add(time.Second),
|
||||
Name: "provider-a", Enabled: false,
|
||||
})
|
||||
if err != nil || !result.Changed {
|
||||
t.Fatalf("SetUpstreamEnabled() = %+v, %v", result, err)
|
||||
}
|
||||
|
||||
first, err := store.Claim(context.Background(), adminstate.ClaimCommand{
|
||||
ConsumerID: "publisher-a", Now: now.Add(2 * time.Second), Limit: 1, Lease: time.Minute,
|
||||
})
|
||||
if err != nil || len(first) != 1 || first[0].ID != 1 {
|
||||
t.Fatalf("Claim(first) = %+v, %v", first, err)
|
||||
}
|
||||
first[0].Payload[0] = 'X'
|
||||
second, err := store.Claim(context.Background(), adminstate.ClaimCommand{
|
||||
ConsumerID: "publisher-b", Now: now.Add(2 * time.Second), Limit: 10, Lease: time.Minute,
|
||||
})
|
||||
if err != nil || len(second) != 1 || second[0].ID != 2 {
|
||||
t.Fatalf("Claim(second) = %+v, %v", second, err)
|
||||
}
|
||||
if err := store.Acknowledge(context.Background(), adminstate.AcknowledgeCommand{
|
||||
ConsumerID: "publisher-b", Now: now.Add(3 * time.Second), EventIDs: []uint64{1},
|
||||
}); !errors.Is(err, adminstate.ErrConflict) {
|
||||
t.Fatalf("Acknowledge(wrong owner) error = %v", err)
|
||||
}
|
||||
|
||||
reclaimed, err := store.Claim(context.Background(), adminstate.ClaimCommand{
|
||||
ConsumerID: "publisher-b", Now: now.Add(2*time.Minute + time.Second), Limit: 10, Lease: time.Minute,
|
||||
})
|
||||
if err != nil || len(reclaimed) != 2 || reclaimed[0].ID != 1 || reclaimed[1].ID != 2 || reclaimed[0].Payload[0] == 'X' {
|
||||
t.Fatalf("Claim(reclaimed) = %+v, %v", reclaimed, err)
|
||||
}
|
||||
if err := store.Acknowledge(context.Background(), adminstate.AcknowledgeCommand{
|
||||
ConsumerID: "publisher-b", Now: now.Add(2*time.Minute + 2*time.Second), EventIDs: []uint64{1, 2},
|
||||
}); err != nil {
|
||||
t.Fatalf("Acknowledge(valid): %v", err)
|
||||
}
|
||||
empty, err := store.Claim(context.Background(), adminstate.ClaimCommand{
|
||||
ConsumerID: "publisher-c", Now: now.Add(4 * time.Minute), Limit: 10, Lease: time.Minute,
|
||||
})
|
||||
if err != nil || len(empty) != 0 {
|
||||
t.Fatalf("Claim(after ACK) = %+v, %v", empty, err)
|
||||
}
|
||||
}
|
||||
|
||||
func runContextContract(t *testing.T, store adminstate.Store) {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
command := configCommand("req-config", "cfg-1", strings.Repeat("a", adminstate.SHA256HexBytes))
|
||||
if _, err := store.CommitConfig(ctx, command); !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("CommitConfig(canceled) error = %v", err)
|
||||
}
|
||||
if _, err := store.Snapshot(ctx); !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("Snapshot(canceled) error = %v", err)
|
||||
}
|
||||
if _, err := store.ReadAudit(ctx, adminstate.AuditQuery{Limit: 1}); !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("ReadAudit(canceled) error = %v", err)
|
||||
}
|
||||
if _, err := store.Claim(ctx, adminstate.ClaimCommand{
|
||||
ConsumerID: "publisher-a", Now: contractNow(), Limit: 1, Lease: time.Second,
|
||||
}); !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("Claim(canceled) error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func configCommand(requestID, version, checksum string) adminstate.CommitConfigCommand {
|
||||
return adminstate.CommitConfigCommand{
|
||||
RequestID: requestID, Actor: contractActor(), OccurredAt: contractNow(),
|
||||
ConfigVersion: version, Checksum: checksum, Source: "configs/proxy-pool.yaml",
|
||||
Upstreams: []adminstate.UpstreamDefinition{
|
||||
{Name: "provider-a", Enabled: true},
|
||||
{Name: "provider-b", Enabled: true},
|
||||
},
|
||||
Routings: []adminstate.RoutingDefinition{{
|
||||
Name: "checkout", Enabled: true, Upstreams: []string{"provider-a", "provider-b"},
|
||||
CurrentUpstream: "provider-a",
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
func contractActor() adminstate.Actor {
|
||||
return adminstate.Actor{ID: "admin-a", SourceIP: "192.0.2.10"}
|
||||
}
|
||||
|
||||
func contractNow() time.Time {
|
||||
return time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC)
|
||||
}
|
||||
|
||||
func commit(t *testing.T, store adminstate.Store, command adminstate.CommitConfigCommand) {
|
||||
t.Helper()
|
||||
if result, err := store.CommitConfig(context.Background(), command); err != nil || !result.Changed {
|
||||
t.Fatalf("CommitConfig() = %+v, %v", result, err)
|
||||
}
|
||||
}
|
||||
|
||||
func upstreamEnabled(snapshot adminstate.Snapshot, name string) bool {
|
||||
for _, upstream := range snapshot.Upstreams {
|
||||
if upstream.Name == name {
|
||||
return upstream.Enabled
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
412
internal/domain/adminstate/memory.go
Normal file
412
internal/domain/adminstate/memory.go
Normal file
@ -0,0 +1,412 @@
|
||||
package adminstate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"math"
|
||||
"net/netip"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
var _ Store = (*MemoryStore)(nil)
|
||||
|
||||
type MemoryStore struct {
|
||||
mu sync.Mutex
|
||||
revision uint64
|
||||
config *ConfigRevision
|
||||
configChecksums map[string]string
|
||||
upstreams map[string]UpstreamState
|
||||
routings map[string]RoutingState
|
||||
audits []AuditRecord
|
||||
events []Event
|
||||
nextAuditID uint64
|
||||
nextEventID uint64
|
||||
}
|
||||
|
||||
func NewMemoryStore() *MemoryStore {
|
||||
return &MemoryStore{
|
||||
configChecksums: make(map[string]string),
|
||||
upstreams: make(map[string]UpstreamState),
|
||||
routings: make(map[string]RoutingState),
|
||||
}
|
||||
}
|
||||
|
||||
func (store *MemoryStore) CommitConfig(ctx context.Context, command CommitConfigCommand) (MutationResult, error) {
|
||||
result := MutationResult{RequestID: command.RequestID}
|
||||
if err := contextError(ctx); err != nil {
|
||||
return result, err
|
||||
}
|
||||
if store == nil || validateCommitConfigCommand(command) != nil {
|
||||
return result, ErrInvalidCommand
|
||||
}
|
||||
command = cloneCommitConfigCommand(command)
|
||||
|
||||
store.mu.Lock()
|
||||
defer store.mu.Unlock()
|
||||
if err := ctx.Err(); err != nil {
|
||||
return result, err
|
||||
}
|
||||
if checksum, exists := store.configChecksums[command.ConfigVersion]; exists {
|
||||
if checksum != command.Checksum || store.config == nil || store.config.ConfigVersion != command.ConfigVersion {
|
||||
return result, ErrConflict
|
||||
}
|
||||
result.Revision = store.revision
|
||||
store.appendAuditLocked(command.RequestID, command.Actor, ActionCommitConfig, "config",
|
||||
command.ConfigVersion, false, store.revision, "", command.OccurredAt)
|
||||
return result, nil
|
||||
}
|
||||
if store.revision == math.MaxUint64 {
|
||||
return result, ErrUnavailable
|
||||
}
|
||||
|
||||
nextRevision := store.revision + 1
|
||||
nextConfig := &ConfigRevision{
|
||||
Revision: nextRevision, ConfigVersion: command.ConfigVersion, Checksum: command.Checksum,
|
||||
Source: command.Source, CreatedAt: command.OccurredAt.UTC(),
|
||||
}
|
||||
nextUpstreams := make(map[string]UpstreamState, len(command.Upstreams))
|
||||
for _, definition := range command.Upstreams {
|
||||
nextUpstreams[definition.Name] = UpstreamState{
|
||||
Name: definition.Name, Enabled: definition.Enabled, Revision: nextRevision,
|
||||
UpdatedAt: command.OccurredAt.UTC(),
|
||||
}
|
||||
}
|
||||
nextRoutings := make(map[string]RoutingState, len(command.Routings))
|
||||
for _, definition := range command.Routings {
|
||||
nextRoutings[definition.Name] = RoutingState{
|
||||
Name: definition.Name, Enabled: definition.Enabled,
|
||||
Upstreams: append([]string(nil), definition.Upstreams...), CurrentUpstream: definition.CurrentUpstream,
|
||||
Revision: nextRevision, UpdatedAt: command.OccurredAt.UTC(),
|
||||
}
|
||||
}
|
||||
payload, err := encodeEventPayload(map[string]any{
|
||||
"configVersion": command.ConfigVersion,
|
||||
"checksum": command.Checksum,
|
||||
"revision": nextRevision,
|
||||
})
|
||||
if err != nil {
|
||||
return result, ErrUnavailable
|
||||
}
|
||||
|
||||
store.revision = nextRevision
|
||||
store.config = nextConfig
|
||||
store.configChecksums[command.ConfigVersion] = command.Checksum
|
||||
store.upstreams = nextUpstreams
|
||||
store.routings = nextRoutings
|
||||
store.appendAuditLocked(command.RequestID, command.Actor, ActionCommitConfig, "config",
|
||||
command.ConfigVersion, true, nextRevision, "", command.OccurredAt)
|
||||
store.appendEventLocked(nextRevision, "config.committed", "config", command.ConfigVersion,
|
||||
payload, command.OccurredAt)
|
||||
return MutationResult{RequestID: command.RequestID, Changed: true, Revision: nextRevision}, nil
|
||||
}
|
||||
|
||||
func (store *MemoryStore) SetUpstreamEnabled(ctx context.Context, command SetUpstreamCommand) (MutationResult, error) {
|
||||
result := MutationResult{RequestID: command.RequestID}
|
||||
if err := contextError(ctx); err != nil {
|
||||
return result, err
|
||||
}
|
||||
if store == nil || validateSetUpstreamCommand(command) != nil {
|
||||
return result, ErrInvalidCommand
|
||||
}
|
||||
|
||||
store.mu.Lock()
|
||||
defer store.mu.Unlock()
|
||||
if err := ctx.Err(); err != nil {
|
||||
return result, err
|
||||
}
|
||||
state, exists := store.upstreams[command.Name]
|
||||
if !exists {
|
||||
return result, ErrNotFound
|
||||
}
|
||||
if state.Enabled == command.Enabled {
|
||||
result.Revision = store.revision
|
||||
store.appendAuditLocked(command.RequestID, command.Actor, ActionSetUpstream, "upstream",
|
||||
command.Name, false, store.revision, "", command.OccurredAt)
|
||||
return result, nil
|
||||
}
|
||||
if store.revision == math.MaxUint64 {
|
||||
return result, ErrUnavailable
|
||||
}
|
||||
|
||||
nextRevision := store.revision + 1
|
||||
payload, err := encodeEventPayload(map[string]any{
|
||||
"enabled": command.Enabled,
|
||||
"name": command.Name,
|
||||
"revision": nextRevision,
|
||||
})
|
||||
if err != nil {
|
||||
return result, ErrUnavailable
|
||||
}
|
||||
state.Enabled = command.Enabled
|
||||
state.Revision = nextRevision
|
||||
state.UpdatedAt = command.OccurredAt.UTC()
|
||||
store.revision = nextRevision
|
||||
store.upstreams[command.Name] = state
|
||||
store.appendAuditLocked(command.RequestID, command.Actor, ActionSetUpstream, "upstream",
|
||||
command.Name, true, nextRevision, "", command.OccurredAt)
|
||||
store.appendEventLocked(nextRevision, "upstream.enabled_changed", "upstream", command.Name,
|
||||
payload, command.OccurredAt)
|
||||
return MutationResult{RequestID: command.RequestID, Changed: true, Revision: nextRevision}, nil
|
||||
}
|
||||
|
||||
func (store *MemoryStore) SwitchRouting(ctx context.Context, command SwitchRoutingCommand) (MutationResult, error) {
|
||||
result := MutationResult{RequestID: command.RequestID}
|
||||
if err := contextError(ctx); err != nil {
|
||||
return result, err
|
||||
}
|
||||
if store == nil || validateSwitchRoutingCommand(command) != 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.Enabled || state.CurrentUpstream != command.ExpectedCurrent {
|
||||
return result, ErrConflict
|
||||
}
|
||||
if !contains(state.Upstreams, command.Target) {
|
||||
return result, ErrInvalidCommand
|
||||
}
|
||||
if state.CurrentUpstream == command.Target {
|
||||
result.Revision = store.revision
|
||||
store.appendAuditLocked(command.RequestID, command.Actor, ActionSwitchRoute, "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.Target,
|
||||
"name": command.Name,
|
||||
"previous": command.ExpectedCurrent,
|
||||
"reason": command.Reason,
|
||||
"revision": nextRevision,
|
||||
})
|
||||
if err != nil {
|
||||
return result, ErrUnavailable
|
||||
}
|
||||
state.CurrentUpstream = command.Target
|
||||
state.Revision = nextRevision
|
||||
state.UpdatedAt = command.OccurredAt.UTC()
|
||||
store.revision = nextRevision
|
||||
store.routings[command.Name] = state
|
||||
store.appendAuditLocked(command.RequestID, command.Actor, ActionSwitchRoute, "routing",
|
||||
command.Name, true, nextRevision, command.Reason, command.OccurredAt)
|
||||
store.appendEventLocked(nextRevision, "routing.switched", "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
|
||||
}
|
||||
if store == nil {
|
||||
return Snapshot{}, ErrInvalidCommand
|
||||
}
|
||||
store.mu.Lock()
|
||||
defer store.mu.Unlock()
|
||||
if err := ctx.Err(); err != nil {
|
||||
return Snapshot{}, err
|
||||
}
|
||||
|
||||
snapshot := Snapshot{Revision: store.revision}
|
||||
if store.config != nil {
|
||||
config := *store.config
|
||||
snapshot.Config = &config
|
||||
}
|
||||
for _, upstream := range store.upstreams {
|
||||
snapshot.Upstreams = append(snapshot.Upstreams, upstream)
|
||||
}
|
||||
for _, routing := range store.routings {
|
||||
routing.Upstreams = append([]string(nil), routing.Upstreams...)
|
||||
snapshot.Routings = append(snapshot.Routings, routing)
|
||||
}
|
||||
sort.Slice(snapshot.Upstreams, func(left, right int) bool {
|
||||
return snapshot.Upstreams[left].Name < snapshot.Upstreams[right].Name
|
||||
})
|
||||
sort.Slice(snapshot.Routings, func(left, right int) bool {
|
||||
return snapshot.Routings[left].Name < snapshot.Routings[right].Name
|
||||
})
|
||||
return cloneSnapshot(snapshot), nil
|
||||
}
|
||||
|
||||
func (store *MemoryStore) ReadAudit(ctx context.Context, query AuditQuery) ([]AuditRecord, error) {
|
||||
if err := contextError(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if store == nil || validateAuditQuery(query) != nil {
|
||||
return nil, ErrInvalidCommand
|
||||
}
|
||||
store.mu.Lock()
|
||||
defer store.mu.Unlock()
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result := make([]AuditRecord, 0, query.Limit)
|
||||
for _, record := range store.audits {
|
||||
if record.ID <= query.AfterID {
|
||||
continue
|
||||
}
|
||||
result = append(result, record)
|
||||
if len(result) == query.Limit {
|
||||
break
|
||||
}
|
||||
}
|
||||
return cloneAuditRecords(result), nil
|
||||
}
|
||||
|
||||
func (store *MemoryStore) Claim(ctx context.Context, command ClaimCommand) ([]Event, error) {
|
||||
if err := contextError(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if store == nil || validateClaimCommand(command) != nil {
|
||||
return nil, ErrInvalidCommand
|
||||
}
|
||||
store.mu.Lock()
|
||||
defer store.mu.Unlock()
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
claimUntil := command.Now.UTC().Add(command.Lease)
|
||||
result := make([]Event, 0, command.Limit)
|
||||
for index := range store.events {
|
||||
event := &store.events[index]
|
||||
if event.PublishedAt != nil || (event.ClaimUntil != nil && command.Now.Before(*event.ClaimUntil)) {
|
||||
continue
|
||||
}
|
||||
event.ClaimedBy = command.ConsumerID
|
||||
event.ClaimUntil = &claimUntil
|
||||
result = append(result, cloneEvent(*event))
|
||||
if len(result) == command.Limit {
|
||||
break
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (store *MemoryStore) Acknowledge(ctx context.Context, command AcknowledgeCommand) error {
|
||||
if err := contextError(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if store == nil || validateAcknowledgeCommand(command) != nil {
|
||||
return ErrInvalidCommand
|
||||
}
|
||||
store.mu.Lock()
|
||||
defer store.mu.Unlock()
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
indexes := make([]int, 0, len(command.EventIDs))
|
||||
for _, eventID := range command.EventIDs {
|
||||
index := store.eventIndexLocked(eventID)
|
||||
if index < 0 {
|
||||
return ErrNotFound
|
||||
}
|
||||
event := store.events[index]
|
||||
if event.PublishedAt != nil || event.ClaimedBy != command.ConsumerID || event.ClaimUntil == nil ||
|
||||
!command.Now.Before(*event.ClaimUntil) {
|
||||
return ErrConflict
|
||||
}
|
||||
indexes = append(indexes, index)
|
||||
}
|
||||
publishedAt := command.Now.UTC()
|
||||
for _, index := range indexes {
|
||||
store.events[index].PublishedAt = &publishedAt
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *MemoryStore) appendAuditLocked(
|
||||
requestID string,
|
||||
actor Actor,
|
||||
action Action,
|
||||
resourceType string,
|
||||
resourceName string,
|
||||
changed bool,
|
||||
revision uint64,
|
||||
reason string,
|
||||
occurredAt time.Time,
|
||||
) {
|
||||
store.nextAuditID++
|
||||
actor = canonicalActor(actor)
|
||||
store.audits = append(store.audits, AuditRecord{
|
||||
ID: store.nextAuditID, RequestID: requestID, Actor: actor, Action: action,
|
||||
ResourceType: resourceType, ResourceName: resourceName, Changed: changed,
|
||||
Revision: revision, Reason: reason, OccurredAt: occurredAt.UTC(),
|
||||
})
|
||||
}
|
||||
|
||||
func (store *MemoryStore) appendEventLocked(
|
||||
revision uint64,
|
||||
eventType string,
|
||||
aggregateType string,
|
||||
aggregateID string,
|
||||
payload json.RawMessage,
|
||||
occurredAt time.Time,
|
||||
) {
|
||||
store.nextEventID++
|
||||
store.events = append(store.events, Event{
|
||||
ID: store.nextEventID, Revision: revision, Type: eventType,
|
||||
AggregateType: aggregateType, AggregateID: aggregateID,
|
||||
Payload: append(json.RawMessage(nil), payload...), OccurredAt: occurredAt.UTC(),
|
||||
})
|
||||
}
|
||||
|
||||
func (store *MemoryStore) eventIndexLocked(eventID uint64) int {
|
||||
index := sort.Search(len(store.events), func(index int) bool {
|
||||
return store.events[index].ID >= eventID
|
||||
})
|
||||
if index >= len(store.events) || store.events[index].ID != eventID {
|
||||
return -1
|
||||
}
|
||||
return index
|
||||
}
|
||||
|
||||
func contextError(ctx context.Context) error {
|
||||
if ctx == nil {
|
||||
return ErrInvalidCommand
|
||||
}
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
func contains(values []string, target string) bool {
|
||||
for _, value := range values {
|
||||
if value == target {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func encodeEventPayload(value any) (json.RawMessage, error) {
|
||||
payload, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return payload, nil
|
||||
}
|
||||
|
||||
func canonicalActor(actor Actor) Actor {
|
||||
if actor.SourceIP == "" {
|
||||
return actor
|
||||
}
|
||||
address, err := netip.ParseAddr(actor.SourceIP)
|
||||
if err == nil {
|
||||
actor.SourceIP = address.Unmap().String()
|
||||
}
|
||||
return actor
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user