fix: serialize persistent profile mutations

This commit is contained in:
proxyctl coordinator
2026-07-17 02:18:58 +08:00
parent f9469cf6a2
commit 80ab995e2e
7 changed files with 422 additions and 85 deletions
+51 -61
View File
@@ -59,99 +59,89 @@ func (a *Application) Snapshot() Snapshot {
func (a *Application) Dispatch(intent Intent) Result {
var notice string
var err error
switch intent.Kind {
case CreateProfile:
notice, err = a.create(intent.Profile)
case ActivateProfile:
notice, err = a.activate(intent.Name)
case UpdateProfile:
notice, err = a.update(intent.Name, intent.Profile)
case DeleteProfile:
notice, err = a.delete(intent.Name, intent.Confirm)
default:
err = errors.New("unsupported Profile action")
next, err := a.store.Mutate(func(current profile.State) (profile.State, error) {
var mutationErr error
switch intent.Kind {
case CreateProfile:
current, notice, mutationErr = create(current, intent.Profile)
case ActivateProfile:
current, notice, mutationErr = activate(current, intent.Name)
case UpdateProfile:
current, notice, mutationErr = update(current, intent.Name, intent.Profile)
case DeleteProfile:
current, notice, mutationErr = deleteProfile(current, intent.Name, intent.Confirm)
default:
mutationErr = errors.New("unsupported Profile action")
}
return current, mutationErr
})
if err == nil {
a.state = next
}
return Result{Snapshot: a.Snapshot(), Notice: notice, Err: err}
}
func (a *Application) create(input profile.Profile) (string, error) {
func create(state profile.State, input profile.Profile) (profile.State, string, error) {
p, err := profile.ValidateAndNormalize(input)
if err != nil {
return "", err
return state, "", err
}
if a.index(p.Name) >= 0 {
return "", fmt.Errorf("a Profile named %q already exists", p.Name)
if index(state, p.Name) >= 0 {
return state, "", fmt.Errorf("a Profile named %q already exists", p.Name)
}
next := cloneState(a.state)
next := cloneState(state)
next.Profiles = append(next.Profiles, p)
if err := a.store.Save(next); err != nil {
return "", err
}
a.state = next
return fmt.Sprintf("Created Profile %q", p.Name), nil
return next, fmt.Sprintf("Created Profile %q", p.Name), nil
}
func (a *Application) activate(name string) (string, error) {
if a.index(name) < 0 {
return "", fmt.Errorf("Profile %q does not exist", name)
func activate(state profile.State, name string) (profile.State, string, error) {
if index(state, name) < 0 {
return state, "", fmt.Errorf("Profile %q does not exist", name)
}
next := cloneState(a.state)
next := cloneState(state)
next.ActiveProfile = name
if err := a.store.Save(next); err != nil {
return "", err
}
a.state = next
return fmt.Sprintf("Activated Profile %q", name), nil
return next, fmt.Sprintf("Activated Profile %q", name), nil
}
func (a *Application) update(originalName string, input profile.Profile) (string, error) {
index := a.index(originalName)
if index < 0 {
return "", fmt.Errorf("Profile %q does not exist", originalName)
func update(state profile.State, originalName string, input profile.Profile) (profile.State, string, error) {
profileIndex := index(state, originalName)
if profileIndex < 0 {
return state, "", fmt.Errorf("Profile %q does not exist", originalName)
}
p, err := profile.ValidateAndNormalize(input)
if err != nil {
return "", err
return state, "", err
}
if other := a.index(p.Name); other >= 0 && other != index {
return "", fmt.Errorf("a Profile named %q already exists", p.Name)
if other := index(state, p.Name); other >= 0 && other != profileIndex {
return state, "", fmt.Errorf("a Profile named %q already exists", p.Name)
}
next := cloneState(a.state)
next.Profiles[index] = p
next := cloneState(state)
next.Profiles[profileIndex] = p
if next.ActiveProfile == originalName {
next.ActiveProfile = p.Name
}
if err := a.store.Save(next); err != nil {
return "", err
}
a.state = next
return fmt.Sprintf("Updated Profile %q", p.Name), nil
return next, fmt.Sprintf("Updated Profile %q", p.Name), nil
}
func (a *Application) delete(name string, confirm bool) (string, error) {
index := a.index(name)
if index < 0 {
return "", fmt.Errorf("Profile %q does not exist", name)
func deleteProfile(state profile.State, name string, confirm bool) (profile.State, string, error) {
profileIndex := index(state, name)
if profileIndex < 0 {
return state, "", fmt.Errorf("Profile %q does not exist", name)
}
if len(a.state.Profiles) == 1 && !confirm {
return "", errors.New("deleting the final Profile requires confirmation")
if len(state.Profiles) == 1 && !confirm {
return state, "", errors.New("deleting the final Profile requires confirmation")
}
next := cloneState(a.state)
next.Profiles = append(next.Profiles[:index], next.Profiles[index+1:]...)
next := cloneState(state)
next.Profiles = append(next.Profiles[:profileIndex], next.Profiles[profileIndex+1:]...)
if next.ActiveProfile == name {
next.ActiveProfile = ""
}
if err := a.store.Save(next); err != nil {
return "", err
}
a.state = next
return fmt.Sprintf("Deleted Profile %q", name), nil
return next, fmt.Sprintf("Deleted Profile %q", name), nil
}
func (a *Application) index(name string) int {
for i := range a.state.Profiles {
if a.state.Profiles[i].Name == name {
func index(state profile.State, name string) int {
for i := range state.Profiles {
if state.Profiles[i].Name == name {
return i
}
}
+59
View File
@@ -5,6 +5,7 @@ import (
"os"
"path/filepath"
"strings"
"sync"
"testing"
"proxyctl/internal/app"
@@ -70,6 +71,64 @@ func TestProfileConfigurationUsesIsolatedXDGPathOwnerOnlyAndAtomicReplacement(t
}
}
func TestConcurrentApplicationsMutateLatestStateWithoutLostUpdates(t *testing.T) {
path := filepath.Join(t.TempDir(), "proxyctl", "profiles.json")
first, err := app.New(profile.NewFileStore(path))
if err != nil {
t.Fatal(err)
}
second, err := app.New(profile.NewFileStore(path))
if err != nil {
t.Fatal(err)
}
start := make(chan struct{})
results := make(chan app.Result, 2)
var workers sync.WaitGroup
for application, name := range map[*app.Application]string{first: "home", second: "office"} {
workers.Add(1)
go func() {
defer workers.Done()
<-start
results <- application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{
Name: name, Mode: profile.SingleMode, SingleURL: "http://proxy.example:8080",
}})
}()
}
close(start)
workers.Wait()
close(results)
for result := range results {
if result.Err != nil {
t.Fatalf("concurrent mutation: %v", result.Err)
}
}
reloaded, err := app.New(profile.NewFileStore(path))
if err != nil {
t.Fatal(err)
}
if got := len(reloaded.Snapshot().Profiles); got != 2 {
t.Fatalf("persisted Profiles = %d, want both concurrent updates", got)
}
assertMode(t, path+".lock", 0o600)
}
func TestLoadSecuresExistingSecretBearingConfiguration(t *testing.T) {
path := filepath.Join(t.TempDir(), "proxyctl", "profiles.json")
if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
t.Fatal(err)
}
contents := `{"profiles":[{"name":"private","mode":"single","singleUrl":"http://alice:secret@proxy.example:8080"}]}`
if err := os.WriteFile(path, []byte(contents), 0o644); err != nil {
t.Fatal(err)
}
if _, err := app.New(profile.NewFileStore(path)); err != nil {
t.Fatalf("load existing configuration: %v", err)
}
assertMode(t, path, 0o600)
}
func assertMode(t *testing.T, path string, want os.FileMode) {
t.Helper()
info, err := os.Stat(path)