feat(natsstore): answer 409 when a management edit loses to another replica

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
 	}

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 {

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())
+	}
+}

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

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

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 {

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 {

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)
+	}
+}

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 {