From 4c7a9c36e80f27e7ece93d8cb03432420db448c8 Mon Sep 17 00:00:00 2001 From: proxyctl coordinator Date: Fri, 17 Jul 2026 03:41:03 +0800 Subject: [PATCH] fix: serialize Docker client recovery mutations --- internal/dockerclient/dockerclient.go | 37 +++-- internal/dockerclient/dockerclient_test.go | 153 ++++++++++++++++++++- internal/securefs/securefs.go | 31 +++++ 3 files changed, 209 insertions(+), 12 deletions(-) diff --git a/internal/dockerclient/dockerclient.go b/internal/dockerclient/dockerclient.go index fb769be..b39fc72 100644 --- a/internal/dockerclient/dockerclient.go +++ b/internal/dockerclient/dockerclient.go @@ -104,6 +104,7 @@ type FileSystem interface { ReadFile(string) ([]byte, error) AtomicWrite(string, []byte, os.FileMode, *Ownership) error Remove(string) error + WithLock(string, *Ownership, func() error) error } type OSFileSystem struct{ TrustedRoots []string } @@ -132,6 +133,9 @@ func (f OSFileSystem) Remove(path string) error { return f.store(pat func (f OSFileSystem) AtomicWrite(path string, data []byte, mode os.FileMode, owner *Ownership) error { return f.store(path, owner).AtomicWrite(path, data, mode) } +func (f OSFileSystem) WithLock(path string, owner *Ownership, action func() error) error { + return f.store(path, owner).WithLock(path, action) +} type priorValue struct { Present bool `json:"present"` @@ -191,6 +195,7 @@ func (a *Adapter) usablePath(path string) bool { } func (a *Adapter) ConfigPath() string { return a.configuration.ConfigFile } func (a *Adapter) RecoveryPath() string { return a.configuration.RecoveryFile } +func (a *Adapter) LockPath() string { return a.configuration.RecoveryFile + ".lock" } func (a *Adapter) Inspect(_ context.Context, active *profile.Profile) target.Inspection { metadata, managed, metadataErr := a.readRecovery() @@ -221,17 +226,29 @@ func (a *Adapter) SetDesired(ctx context.Context, enabled bool, active *profile. if enabled && active == nil { return target.OperationResult{Inspection: a.Inspect(ctx, active), Err: errors.New("no Active Profile; open Profile management to create and activate one")} } - if enabled && a.Inspect(ctx, active).State == target.Enabled { - return target.OperationResult{Inspection: a.Inspect(ctx, active)} - } - var err error - if enabled { - err = a.enable(active) - } else { - err = a.disable() - } - inspection := a.Inspect(ctx, active) + + var inspection target.Inspection + err := a.files.WithLock(a.LockPath(), a.ownership(), func() error { + if enabled { + inspection = a.Inspect(ctx, active) + if inspection.State == target.Enabled { + return nil + } + if err := a.enable(active); err != nil { + inspection = a.Inspect(ctx, active) + return err + } + } else if err := a.disable(); err != nil { + inspection = a.Inspect(ctx, active) + return err + } + inspection = a.Inspect(ctx, active) + return nil + }) if err != nil { + if inspection.Evidence == "" { + inspection = a.Inspect(ctx, active) + } err = redactError(err, active) } return target.OperationResult{Inspection: inspection, Err: err} diff --git a/internal/dockerclient/dockerclient_test.go b/internal/dockerclient/dockerclient_test.go index aa5c4c8..3236b02 100644 --- a/internal/dockerclient/dockerclient_test.go +++ b/internal/dockerclient/dockerclient_test.go @@ -5,9 +5,13 @@ import ( "encoding/json" "errors" "os" + "os/exec" "path/filepath" "strings" + "sync" + "syscall" "testing" + "time" "proxyctl/internal/dockerclient" "proxyctl/internal/profile" @@ -239,6 +243,92 @@ func TestUnmergeableJSONPreventsMutation(t *testing.T) { } } +func TestConcurrentMutationsSerializeRecoveryAndConfiguration(t *testing.T) { + files := &interleavingFS{FileSystem: dockerclient.OSFileSystem{}, firstRecovery: make(chan struct{}), secondRecovery: make(chan struct{}), secondConfig: make(chan struct{})} + a := newAdapterWithFS(t, files) + first := *activeProfile() + first.Name = "first" + first.HTTPURL = "http://first.example:8080" + second := *activeProfile() + second.Name = "second" + second.HTTPURL = "http://second.example:8080" + + firstDone := make(chan target.OperationResult, 1) + go func() { firstDone <- a.SetDesired(context.Background(), true, &first) }() + select { + case <-files.firstRecovery: + case <-time.After(5 * time.Second): + t.Fatal("first mutation did not reach recovery write") + } + secondDone := make(chan target.OperationResult, 1) + go func() { secondDone <- a.SetDesired(context.Background(), true, &second) }() + + for name, done := range map[string]<-chan target.OperationResult{"first": firstDone, "second": secondDone} { + select { + case result := <-done: + if result.Err != nil { + t.Fatalf("%s mutation: %v", name, result.Err) + } + case <-time.After(5 * time.Second): + t.Fatalf("%s mutation did not complete", name) + } + } + if got := a.Inspect(context.Background(), &second); got.State != target.Enabled { + t.Fatalf("final state = %s (%s), want second mutation consistently enabled", got.State, got.Evidence) + } +} + +func TestDockerClientLockSerializesAcrossProcesses(t *testing.T) { + if root := os.Getenv("PROXYCTL_DOCKER_LOCK_HELPER_ROOT"); root != "" { + a := dockerclient.New(dockerclient.Configuration{ + ConfigFile: filepath.Join(root, ".docker", "config.json"), RecoveryFile: filepath.Join(root, ".state", "proxyctl", "docker-containers.json"), + }, dockerclient.OSFileSystem{}) + if result := a.SetDesired(context.Background(), true, activeProfile()); result.Err != nil { + t.Fatal(result.Err) + } + return + } + + a := newAdapter(t, nil) + if err := os.MkdirAll(filepath.Dir(a.LockPath()), 0o700); err != nil { + t.Fatal(err) + } + lock, err := os.OpenFile(a.LockPath(), os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + t.Fatal(err) + } + defer lock.Close() + if err := syscall.Flock(int(lock.Fd()), syscall.LOCK_EX); err != nil { + t.Fatal(err) + } + + root := filepath.Dir(filepath.Dir(filepath.Dir(a.RecoveryPath()))) + command := exec.Command(os.Args[0], "-test.run=^TestDockerClientLockSerializesAcrossProcesses$") + command.Env = append(os.Environ(), "PROXYCTL_DOCKER_LOCK_HELPER_ROOT="+root) + done := make(chan error, 1) + go func() { done <- command.Run() }() + select { + case err := <-done: + t.Fatalf("child mutation did not wait for Docker client lock: %v", err) + case <-time.After(100 * time.Millisecond): + } + if err := syscall.Flock(int(lock.Fd()), syscall.LOCK_UN); err != nil { + t.Fatal(err) + } + select { + case err := <-done: + if err != nil { + t.Fatalf("child mutation after unlock: %v", err) + } + case <-time.After(5 * time.Second): + _ = command.Process.Kill() + t.Fatal("child mutation remained blocked after unlock") + } + if got := a.Inspect(context.Background(), activeProfile()); got.State != target.Enabled { + t.Fatalf("state after serialized child mutation = %s (%s)", got.State, got.Evidence) + } +} + func TestWritesAreOwnerOnlyAtomicAndIdempotent(t *testing.T) { files := &countingFS{FileSystem: dockerclient.OSFileSystem{}} a := newAdapterWithFS(t, files) @@ -252,7 +342,7 @@ func TestWritesAreOwnerOnlyAtomicAndIdempotent(t *testing.T) { if files.writes != writes { t.Fatalf("idempotent enable performed %d additional writes", files.writes-writes) } - for _, path := range []string{a.ConfigPath(), a.RecoveryPath()} { + for _, path := range []string{a.ConfigPath(), a.RecoveryPath(), a.LockPath()} { info, err := os.Stat(path) if err != nil { t.Fatal(err) @@ -274,7 +364,7 @@ func TestOwnedWritesUseConfigurationOwner(t *testing.T) { if got := a.SetDesired(context.Background(), true, activeProfile()); got.Err != nil { t.Fatal(got.Err) } - if len(files.owners) != 2 { + if len(files.owners) != 3 { t.Fatalf("owners = %#v", files.owners) } for _, owner := range files.owners { @@ -448,6 +538,60 @@ func (f *countingFS) AtomicWrite(path string, data []byte, mode os.FileMode, own return f.FileSystem.AtomicWrite(path, data, mode, owner) } +type interleavingFS struct { + dockerclient.FileSystem + mu sync.Mutex + recoveryWrites, configWrites int + firstRecovery, secondRecovery, secondConfig chan struct{} +} + +func (f *interleavingFS) AtomicWrite(path string, data []byte, mode os.FileMode, owner *dockerclient.Ownership) error { + f.mu.Lock() + isRecovery := strings.HasSuffix(path, "docker-containers.json") + if isRecovery { + f.recoveryWrites++ + call := f.recoveryWrites + if call == 1 { + close(f.firstRecovery) + } + } else { + f.configWrites++ + } + recoveryCall, configCall := f.recoveryWrites, f.configWrites + f.mu.Unlock() + + if isRecovery && recoveryCall == 1 { + select { + case <-f.secondRecovery: + case <-time.After(100 * time.Millisecond): + } + err := f.FileSystem.AtomicWrite(path, data, mode, owner) + return err + } + if isRecovery && recoveryCall == 2 { + err := f.FileSystem.AtomicWrite(path, data, mode, owner) + close(f.secondRecovery) + return err + } + if !isRecovery && configCall == 1 { + select { + case <-f.secondConfig: + case <-time.After(100 * time.Millisecond): + } + return f.FileSystem.AtomicWrite(path, data, mode, owner) + } + if !isRecovery && configCall == 2 { + err := f.FileSystem.AtomicWrite(path, data, mode, owner) + close(f.secondConfig) + return err + } + return f.FileSystem.AtomicWrite(path, data, mode, owner) +} + +func (f *interleavingFS) WithLock(path string, owner *dockerclient.Ownership, action func() error) error { + return f.FileSystem.WithLock(path, owner, action) +} + type ownershipFS struct { dockerclient.FileSystem owners []*dockerclient.Ownership @@ -457,3 +601,8 @@ func (f *ownershipFS) AtomicWrite(path string, data []byte, mode os.FileMode, ow f.owners = append(f.owners, owner) return f.FileSystem.AtomicWrite(path, data, mode, nil) } + +func (f *ownershipFS) WithLock(path string, owner *dockerclient.Ownership, action func() error) error { + f.owners = append(f.owners, owner) + return f.FileSystem.WithLock(path, nil, action) +} diff --git a/internal/securefs/securefs.go b/internal/securefs/securefs.go index ad882ac..471b3f5 100644 --- a/internal/securefs/securefs.go +++ b/internal/securefs/securefs.go @@ -146,6 +146,37 @@ func (s Store) AtomicWrite(path string, data []byte, mode os.FileMode) error { return directory.Sync() } +// WithLock serializes an action across processes using a descriptor-anchored, +// owner-restricted advisory lock file. +func (s Store) WithLock(path string, action func() error) error { + directory, base, err := s.openParent(path, true) + if err != nil { + return err + } + defer directory.Close() + lock, err := OpenAt(directory, base, unix.O_CREAT|unix.O_RDWR, 0o600) + if err != nil { + return err + } + defer lock.Close() + if err := regular(lock); err != nil { + return err + } + if s.Owner != nil && (s.Owner.UID != os.Geteuid() || s.Owner.GID != os.Getegid()) { + if err := lock.Chown(s.Owner.UID, s.Owner.GID); err != nil { + return err + } + } + if err := lock.Chmod(0o600); err != nil { + return err + } + if err := unix.Flock(int(lock.Fd()), unix.LOCK_EX); err != nil { + return err + } + defer func() { _ = unix.Flock(int(lock.Fd()), unix.LOCK_UN) }() + return action() +} + func (s Store) Remove(path string) error { directory, base, err := s.openParent(path, false) if err != nil {