From 550a1ad148e8dd1d4f66b5e7b707973a18f35239 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Sat, 7 Nov 2026 21:06:04 -0500 Subject: [PATCH] feat(natsstore): answer 409 when a management edit loses to another replica With live sync a management edit can race an edit made on another replica. The config used to be stored later by the file watcher, so the caller got a 200 for a change that was then rejected or overwritten. Commit the config synchronously for shared stores, report the lost compare-and-swap as a conflict, and adopt the winning config so the retry builds on it. Stores that are not shared are unaffected. Signed-off-by: Yordis Prieto --- .../handlers/management/auth_files_crud.go | 3 + .../api/handlers/management/config_basic.go | 3 + .../management/config_conflict_test.go | 94 +++++++++++++++++++ internal/api/handlers/management/config_v8.go | 3 + internal/api/handlers/management/handler.go | 31 ++++++ internal/store/natsstore/kv.go | 7 +- internal/store/natsstore/store.go | 38 ++++++++ internal/store/natsstore/store_test.go | 40 ++++++++ sdk/cliproxy/auth/store.go | 9 +- 9 files changed, 225 insertions(+), 3 deletions(-) create mode 100644 internal/api/handlers/management/config_conflict_test.go diff --git a/internal/api/handlers/management/auth_files_crud.go b/internal/api/handlers/management/auth_files_crud.go index ddecf8cee35..4f6f39e178e 100644 --- a/internal/api/handlers/management/auth_files_crud.go +++ b/internal/api/handlers/management/auth_files_crud.go @@ -132,6 +132,9 @@ func (h *Handler) UploadAuthFile(c *gin.Context) { if errors.Is(err, errInvalidAuthFile) { status = http.StatusBadRequest } + if errors.Is(err, coreauth.ErrStoreConflict) { + status = http.StatusConflict + } c.JSON(status, gin.H{"error": err.Error()}) return } diff --git a/internal/api/handlers/management/config_basic.go b/internal/api/handlers/management/config_basic.go index a6c02c4d60d..d8a2a8a555c 100644 --- a/internal/api/handlers/management/config_basic.go +++ b/internal/api/handlers/management/config_basic.go @@ -170,6 +170,9 @@ func (h *Handler) PutConfigYAML(c *gin.Context) { c.JSON(http.StatusInternalServerError, gin.H{"error": "write_failed", "message": "failed to write config"}) return } + if !h.commitConfig(c) { + return + } // Reload into handler to keep memory in sync newCfg, err := config.LoadConfig(h.configFilePath) if err != nil { diff --git a/internal/api/handlers/management/config_conflict_test.go b/internal/api/handlers/management/config_conflict_test.go new file mode 100644 index 00000000000..85341944868 --- /dev/null +++ b/internal/api/handlers/management/config_conflict_test.go @@ -0,0 +1,94 @@ +package management + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/gin-gonic/gin" + "github.com/router-for-me/CLIProxyAPI/v8/internal/config" + coreauth "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/auth" +) + +// committingStore is a shared token store whose config commit result is scripted. +type committingStore struct { + memoryAuthStore + commitErr error + commits int +} + +func (s *committingStore) CommitConfig(context.Context) error { + s.commits++ + return s.commitErr +} + +func newCommitHandler(t *testing.T, store coreauth.Store) (*Handler, *gin.Engine) { + t.Helper() + gin.SetMode(gin.TestMode) + path := filepath.Join(t.TempDir(), "config.yaml") + if err := os.WriteFile(path, []byte("config-version: 8\nserver: {port: 8317}\n"), 0o600); err != nil { + t.Fatal(err) + } + cfg, err := config.LoadConfig(path) + if err != nil { + t.Fatal(err) + } + h := &Handler{cfg: cfg, configFilePath: path, tokenStore: store} + router := gin.New() + router.PUT("/debug", h.PutDebug) + router.PUT("/config.yaml", h.PutConfigYAML) + return h, router +} + +func putJSON(router *gin.Engine, url, body string) *httptest.ResponseRecorder { + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest(http.MethodPut, url, strings.NewReader(body))) + return rec +} + +func TestConfigWriteAnswers409OnSharedStoreConflict(t *testing.T) { + conflict := fmt.Errorf("commit: %w", coreauth.ErrStoreConflict) + store := &committingStore{commitErr: conflict} + _, router := newCommitHandler(t, store) + + if rec := putJSON(router, "/debug", `{"value":true}`); rec.Code != http.StatusConflict { + t.Fatalf("scalar edit status = %d, want 409, body=%s", rec.Code, rec.Body.String()) + } + if rec := putJSON(router, "/config.yaml", "port: 8317\n"); rec.Code != http.StatusConflict { + t.Fatalf("yaml put status = %d, want 409, body=%s", rec.Code, rec.Body.String()) + } + if store.commits != 2 { + t.Fatalf("commits = %d, want 2", store.commits) + } +} + +func TestConfigWriteSucceedsWhenSharedStoreCommits(t *testing.T) { + store := &committingStore{} + _, router := newCommitHandler(t, store) + if rec := putJSON(router, "/debug", `{"value":true}`); rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200, body=%s", rec.Code, rec.Body.String()) + } + if store.commits != 1 { + t.Fatalf("commits = %d, want 1", store.commits) + } +} + +func TestConfigWriteReportsOtherCommitFailuresAs500(t *testing.T) { + _, router := newCommitHandler(t, &committingStore{commitErr: errors.New("bucket unavailable")}) + if rec := putJSON(router, "/debug", `{"value":true}`); rec.Code != http.StatusInternalServerError { + t.Fatalf("status = %d, want 500, body=%s", rec.Code, rec.Body.String()) + } +} + +func TestConfigWriteIgnoresStoresThatAreNotShared(t *testing.T) { + _, router := newCommitHandler(t, &memoryAuthStore{}) + if rec := putJSON(router, "/debug", `{"value":true}`); rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200, body=%s", rec.Code, rec.Body.String()) + } +} diff --git a/internal/api/handlers/management/config_v8.go b/internal/api/handlers/management/config_v8.go index 5f944095803..a4246a4675b 100644 --- a/internal/api/handlers/management/config_v8.go +++ b/internal/api/handlers/management/config_v8.go @@ -198,6 +198,9 @@ func (h *Handler) ConfigV8(c *gin.Context) { c.JSON(http.StatusInternalServerError, gin.H{"error": "write_failed", "message": err.Error()}) return } + if !h.commitConfig(c) { + return + } // Runtime-only ownership state is never sourced from an uploaded YAML document. next.Home = h.cfg.Home // WriteConfig may normalize the bytes, so report the version actually on diff --git a/internal/api/handlers/management/handler.go b/internal/api/handlers/management/handler.go index 7e8db2f76e6..4e7b12d6ff0 100644 --- a/internal/api/handlers/management/handler.go +++ b/internal/api/handlers/management/handler.go @@ -5,6 +5,7 @@ package management import ( "context" "crypto/subtle" + "errors" "fmt" "net/http" "os" @@ -457,6 +458,33 @@ func (h *Handler) AuthenticateManagementKey(clientIP string, localClient bool, p return true, 0, "" } +// configCommitter is implemented by token stores shared between replicas that +// can store the config file synchronously and detect a concurrent change. +type configCommitter interface { + CommitConfig(ctx context.Context) error +} + +// commitConfig stores the config file just written in a shared backend. It +// answers 409 when another replica changed the config first, in which case the +// file now holds that config, and 500 on other failures. It reports whether +// the caller may go on, and does nothing for stores that are not shared. +func (h *Handler) commitConfig(c *gin.Context) bool { + committer, ok := h.tokenStore.(configCommitter) + if !ok { + return true + } + err := committer.CommitConfig(c.Request.Context()) + if err == nil { + return true + } + if errors.Is(err, coreauth.ErrStoreConflict) { + c.JSON(http.StatusConflict, gin.H{"error": "conflict", "message": "the config was changed by another replica, reload it and retry"}) + return false + } + c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("failed to store config: %v", err)}) + return false +} + // persistLocked saves the current in-memory config to disk. // It expects the caller to hold h.mu. func (h *Handler) persistLocked(c *gin.Context) bool { @@ -465,6 +493,9 @@ func (h *Handler) persistLocked(c *gin.Context) bool { c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("failed to save config: %v", err)}) return false } + if !h.commitConfig(c) { + return false + } snapshot := h.reloadSnapshotConfigLocked() c.JSON(http.StatusOK, gin.H{"status": "ok"}) var reqCtx context.Context diff --git a/internal/store/natsstore/kv.go b/internal/store/natsstore/kv.go index a0d81214e5b..d4d81545896 100644 --- a/internal/store/natsstore/kv.go +++ b/internal/store/natsstore/kv.go @@ -7,6 +7,7 @@ import ( "time" "github.com/nats-io/nats.go/jetstream" + coreauth "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/auth" ) var ( @@ -27,8 +28,10 @@ func (e *ConflictError) Error() string { return fmt.Sprintf("revision conflict on %s in bucket %s", e.ID, e.Bucket) } -// Is makes errors.Is(err, ErrConflict) match. -func (e *ConflictError) Is(target error) bool { return target == ErrConflict } +// Is makes errors.Is match ErrConflict and the SDK-level auth.ErrStoreConflict. +func (e *ConflictError) Is(target error) bool { + return target == ErrConflict || target == coreauth.ErrStoreConflict +} // Entry is one live value with the revision a later write must present. type Entry struct { diff --git a/internal/store/natsstore/store.go b/internal/store/natsstore/store.go index f6340023c2a..1f7ed4911aa 100644 --- a/internal/store/natsstore/store.go +++ b/internal/store/natsstore/store.go @@ -294,6 +294,44 @@ func (s *Store) PersistConfig(ctx context.Context) error { return nil } +// CommitConfig stores the spool config file now and reports a lost +// compare-and-swap as a ConflictError, for callers that must tell their user +// the edit did not land. On a conflict the spool adopts the config that won, so +// the next edit builds on it. +func (s *Store) CommitConfig(ctx context.Context) error { + s.configMu.Lock() + defer s.configMu.Unlock() + data, err := os.ReadFile(s.configPath) + if err != nil { + return fmt.Errorf("nats store: read config file: %w", err) + } + if len(bytes.TrimSpace(data)) == 0 { + return nil + } + s.mu.Lock() + unchanged := bytes.Equal(s.configData, []byte(normalizeLineEndings(string(data)))) + s.mu.Unlock() + if unchanged { + return nil + } + _, err = s.config.Write(ctx, data) + if err == nil { + s.rememberConfig(data) + return nil + } + if !errors.Is(err, ErrConflict) { + return fmt.Errorf("nats store: commit config: %w", err) + } + remote, _, errRead := s.config.Read(ctx) + if errRead != nil { + return fmt.Errorf("nats store: adopt the winning config: %w", errRead) + } + if errWrite := s.writeConfigFile(remote); errWrite != nil { + return errWrite + } + return err +} + // PersistAuthFiles stores the given spool credential files in the bucket, and // removes the credentials whose files are gone or empty. func (s *Store) PersistAuthFiles(ctx context.Context, _ string, paths ...string) error { diff --git a/internal/store/natsstore/store_test.go b/internal/store/natsstore/store_test.go index 2c8bd7b1447..9e697c402d3 100644 --- a/internal/store/natsstore/store_test.go +++ b/internal/store/natsstore/store_test.go @@ -9,6 +9,7 @@ import ( "time" "github.com/nats-io/nats.go" + coreauth "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/auth" "github.com/router-for-me/CLIProxyAPI/v8/internal/store/natsstore/natstest" ) @@ -327,3 +328,42 @@ func TestStoreWritesCarryTheReplicaOrigin(t *testing.T) { t.Fatalf("stored payload = %q", got) } } + +func TestStoreCommitConfigReportsConflictAndAdoptsWinner(t *testing.T) { + f := newStoreFixture(t) + ctx := testContext(t) + winner, loser := f.open(), f.open() + for _, s := range []*Store{winner, loser} { + if err := s.Bootstrap(ctx, f.example); err != nil { + t.Fatal(err) + } + } + if err := os.WriteFile(winner.ConfigPath(), []byte("port: 9000\n"), 0o600); err != nil { + t.Fatal(err) + } + if err := winner.CommitConfig(ctx); err != nil { + t.Fatal(err) + } + + if err := os.WriteFile(loser.ConfigPath(), []byte("port: 1\n"), 0o600); err != nil { + t.Fatal(err) + } + err := loser.CommitConfig(ctx) + if !errors.Is(err, ErrConflict) || !errors.Is(err, coreauth.ErrStoreConflict) { + t.Fatalf("stale commit = %v, want conflict", err) + } + if got := readFile(t, loser.ConfigPath()); string(got) != "port: 9000\n" { + t.Fatalf("spool after conflict = %q, want the winning config", got) + } + + // The next edit builds on the adopted config and lands. + if err := os.WriteFile(loser.ConfigPath(), []byte("port: 2\n"), 0o600); err != nil { + t.Fatal(err) + } + if err := loser.CommitConfig(ctx); err != nil { + t.Fatalf("retry after adopting = %v", err) + } + if data, _, err := f.open().config.Read(ctx); err != nil || string(data) != "port: 2\n" { + t.Fatalf("stored config = %q, %v", data, err) + } +} diff --git a/sdk/cliproxy/auth/store.go b/sdk/cliproxy/auth/store.go index a903a46bf3a..fcf12cbc70c 100644 --- a/sdk/cliproxy/auth/store.go +++ b/sdk/cliproxy/auth/store.go @@ -1,6 +1,13 @@ package auth -import "context" +import ( + "context" + "errors" +) + +// ErrStoreConflict is matched by the error a store returns when a write lost to +// a concurrent change made by another replica sharing the backend. +var ErrStoreConflict = errors.New("store revision conflict") // Store abstracts persistence of Auth state across restarts. type Store interface {