From 606217d900e5e827d4f0e6e47634e8580e113204 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Sat, 7 Nov 2026 20:40:52 -0500 Subject: [PATCH] feat(natsstore): store the config file in the config bucket Operator edits are never merged, so a replica holding a stale config must learn it lost instead of silently replacing a newer one. Checking each write against the revision it last read surfaces that as a conflict. Signed-off-by: Yordis Prieto --- internal/store/natsstore/configstore.go | 77 ++++++++++++++ internal/store/natsstore/configstore_test.go | 101 +++++++++++++++++++ 2 files changed, 178 insertions(+) create mode 100644 internal/store/natsstore/configstore.go create mode 100644 internal/store/natsstore/configstore_test.go diff --git a/internal/store/natsstore/configstore.go b/internal/store/natsstore/configstore.go new file mode 100644 index 00000000000..cff7c47a85f --- /dev/null +++ b/internal/store/natsstore/configstore.go @@ -0,0 +1,77 @@ +package natsstore + +import ( + "context" + "errors" + "strings" + "sync" +) + +// configID is the single key holding config.yaml in the config bucket. +const configID = "config.yaml" + +// ConfigStore keeps the configuration file as one value in the config bucket. +// Every write is checked against the revision this store last read or wrote, +// and a lost race is returned as a conflict instead of overwriting the newer +// config, since operator edits are never merged. +type ConfigStore struct { + kv *KV + + mu sync.Mutex + revision uint64 +} + +// NewConfigStore creates a store over the config bucket. +func NewConfigStore(kv *KV) *ConfigStore { return &ConfigStore{kv: kv} } + +// Read returns the stored config and remembers its revision for the next +// write. It returns ErrNotFound when no config was stored yet. +func (s *ConfigStore) Read(ctx context.Context) ([]byte, uint64, error) { + entry, err := s.kv.Get(ctx, configID) + s.mu.Lock() + defer s.mu.Unlock() + if err != nil { + if errors.Is(err, ErrNotFound) { + s.revision = 0 + } + return nil, 0, err + } + s.revision = entry.Revision + return entry.Value, entry.Revision, nil +} + +// Revision returns the revision the next write is checked against, zero when +// no config is known to exist. +func (s *ConfigStore) Revision() uint64 { + s.mu.Lock() + defer s.mu.Unlock() + return s.revision +} + +// Write stores the config. It creates the key when no config is known to +// exist and otherwise replaces the revision last seen, returning a +// ConflictError when another writer changed it since. +func (s *ConfigStore) Write(ctx context.Context, data []byte) (uint64, error) { + data = []byte(normalizeLineEndings(string(data))) + s.mu.Lock() + defer s.mu.Unlock() + var ( + rev uint64 + err error + ) + if s.revision == 0 { + rev, err = s.kv.Create(ctx, configID, data) + } else { + rev, err = s.kv.Update(ctx, configID, data, s.revision) + } + if err != nil { + return 0, err + } + s.revision = rev + return rev, nil +} + +func normalizeLineEndings(s string) string { + s = strings.ReplaceAll(s, "\r\n", "\n") + return strings.ReplaceAll(s, "\r", "\n") +} diff --git a/internal/store/natsstore/configstore_test.go b/internal/store/natsstore/configstore_test.go new file mode 100644 index 00000000000..d98fc096807 --- /dev/null +++ b/internal/store/natsstore/configstore_test.go @@ -0,0 +1,101 @@ +package natsstore + +import ( + "errors" + "testing" + "time" + + "github.com/router-for-me/CLIProxyAPI/v8/internal/store/natsstore/natstest" +) + +func newConfigStores(t *testing.T, n int) []*ConfigStore { + t.Helper() + srv := natstest.Start(t) + nc, err := connect(t, map[string]string{EnvURL: srv.URL}, 5*time.Second) + if err != nil { + t.Fatal(err) + } + settings := autoSettings("cfgtest") + boot, err := Bootstrap(testContext(t), nc, settings) + if err != nil { + t.Fatal(err) + } + kv := NewKV(settings.BucketName(BucketConfig), boot.Bucket(settings, BucketConfig)) + stores := make([]*ConfigStore, n) + for i := range stores { + stores[i] = NewConfigStore(kv) + } + return stores +} + +func TestConfigStoreReadWrite(t *testing.T) { + store := newConfigStores(t, 1)[0] + ctx := testContext(t) + + if _, _, err := store.Read(ctx); !errors.Is(err, ErrNotFound) { + t.Fatalf("Read before any write = %v, want ErrNotFound", err) + } + rev1, err := store.Write(ctx, []byte("port: 8317\r\nhost: \"\"\r\n")) + if err != nil { + t.Fatal(err) + } + data, rev, err := store.Read(ctx) + if err != nil || rev != rev1 || string(data) != "port: 8317\nhost: \"\"\n" { + t.Fatalf("Read = %q, %d, %v", data, rev, err) + } + rev2, err := store.Write(ctx, []byte("port: 9000\n")) + if err != nil || rev2 <= rev1 || store.Revision() != rev2 { + t.Fatalf("second Write = %d, %v (revision %d)", rev2, err, store.Revision()) + } +} + +func TestConfigStoreConflictsInsteadOfOverwriting(t *testing.T) { + stores := newConfigStores(t, 2) + a, b := stores[0], stores[1] + ctx := testContext(t) + + if _, err := a.Write(ctx, []byte("v: 1\n")); err != nil { + t.Fatal(err) + } + if _, _, err := b.Read(ctx); err != nil { + t.Fatal(err) + } + if _, err := b.Write(ctx, []byte("v: 2\n")); err != nil { + t.Fatal(err) + } + if _, err := a.Write(ctx, []byte("v: 3\n")); !errors.Is(err, ErrConflict) { + t.Fatalf("stale Write = %v, want conflict", err) + } + data, _, err := a.Read(ctx) + if err != nil || string(data) != "v: 2\n" { + t.Fatalf("Read = %q, %v; the newer config must survive", data, err) + } + if _, err := a.Write(ctx, []byte("v: 3\n")); err != nil { + t.Fatalf("Write after re-reading: %v", err) + } +} + +func TestConfigStoreFirstWriteRaceConflicts(t *testing.T) { + stores := newConfigStores(t, 2) + ctx := testContext(t) + + if _, err := stores[0].Write(ctx, []byte("a: 1\n")); err != nil { + t.Fatal(err) + } + if _, err := stores[1].Write(ctx, []byte("b: 1\n")); !errors.Is(err, ErrConflict) { + t.Fatalf("second creator = %v, want conflict", err) + } +} + +func TestConfigStoreStoresEmptyConfig(t *testing.T) { + store := newConfigStores(t, 1)[0] + ctx := testContext(t) + + if _, err := store.Write(ctx, nil); err != nil { + t.Fatal(err) + } + data, _, err := store.Read(ctx) + if err != nil || len(data) != 0 { + t.Fatalf("Read = %q, %v", data, err) + } +}