From 3075fa796195489559f34e5ffe9dae4a40d5beda Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Sat, 7 Nov 2026 21:08:40 -0500 Subject: [PATCH] test(natsstore): cover two replicas sharing one bucket end to end The pieces of live sync are tested alone, so nothing showed that they hold together behind the management API. Drive edits and credential changes through one replica, assert they reach the other without bouncing back, and that an edit from a stale replica gets a 409. Signed-off-by: Yordis Prieto --- .../handlers/management/natsstore_e2e_test.go | 234 ++++++++++++++++++ 1 file changed, 234 insertions(+) create mode 100644 internal/api/handlers/management/natsstore_e2e_test.go diff --git a/internal/api/handlers/management/natsstore_e2e_test.go b/internal/api/handlers/management/natsstore_e2e_test.go new file mode 100644 index 00000000000..df39c443cfd --- /dev/null +++ b/internal/api/handlers/management/natsstore_e2e_test.go @@ -0,0 +1,234 @@ +package management + +import ( + "context" + "net/http" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/fsnotify/fsnotify" + "github.com/gin-gonic/gin" + "github.com/nats-io/nats.go/jetstream" + "github.com/router-for-me/CLIProxyAPI/v8/internal/config" + "github.com/router-for-me/CLIProxyAPI/v8/internal/store/natsstore" + "github.com/router-for-me/CLIProxyAPI/v8/internal/store/natsstore/natstest" + coreauth "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/auth" +) + +// replica is one process-like participant: its own spool, store and handler. +type replica struct { + store *natsstore.Store + router *gin.Engine + config jetstream.KeyValue +} + +type natsCluster struct { + t *testing.T + url string + example string + settings natsstore.StoreSettings +} + +func newNatsCluster(t *testing.T) *natsCluster { + t.Helper() + gin.SetMode(gin.TestMode) + srv := natstest.Start(t) + example, err := filepath.Abs(filepath.Join("..", "..", "..", "..", "config.example.yaml")) + if err != nil { + t.Fatal(err) + } + settings := natsstore.DefaultStoreSettings() + settings.Prefix = "e2e" + settings.Create = natsstore.CreateAuto + return &natsCluster{t: t, url: srv.URL, example: example, settings: settings} +} + +// join starts a replica. A replica that does not sync only sees the buckets +// as they were when it joined, which makes it stale on purpose. +func (n *natsCluster) join(sync bool) *replica { + n.t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + env := func(key string) (string, bool) { + if key == natsstore.EnvURL { + return n.url, true + } + return "", false + } + nc, err := natsstore.Connect(ctx, natsstore.SettingsFromEnv(env).Resolve(), env, 5*time.Second) + if err != nil { + n.t.Fatal(err) + } + boot, err := natsstore.Bootstrap(ctx, nc, n.settings) + if err != nil { + nc.Close() + n.t.Fatal(err) + } + store, err := natsstore.NewStore(nc, boot, n.settings, n.t.TempDir()) + if err != nil { + nc.Close() + n.t.Fatal(err) + } + if err := store.Bootstrap(ctx, n.example); err != nil { + _ = store.Close() + n.t.Fatal(err) + } + if sync { + store.StartSync(context.Background()) + } + n.t.Cleanup(func() { _ = store.Close() }) + + cfg, err := config.LoadConfig(store.ConfigPath()) + if err != nil { + n.t.Fatal(err) + } + h := &Handler{cfg: cfg, configFilePath: store.ConfigPath(), tokenStore: store} + router := gin.New() + router.PUT("/debug", h.PutDebug) + return &replica{ + store: store, + router: router, + config: boot.Bucket(n.settings, natsstore.BucketConfig), + } +} + +func (r *replica) setDebug(value string) int { + return putJSON(r.router, "/debug", `{"value":`+value+`}`).Code +} + +func (r *replica) configText() string { + data, err := os.ReadFile(r.store.ConfigPath()) + if err != nil { + return "" + } + return string(data) +} + +func (r *replica) debugOn() bool { + for _, line := range strings.Split(r.configText(), "\n") { + if strings.HasPrefix(strings.TrimSpace(line), "debug:") { + return strings.Contains(line, "true") + } + } + return false +} + +func (r *replica) configRevision(t *testing.T) uint64 { + t.Helper() + key, err := natsstore.EncodeKey("config.yaml") + if err != nil { + t.Fatal(err) + } + entry, err := r.config.Get(context.Background(), key) + if err != nil { + t.Fatal(err) + } + return entry.Revision() +} + +func eventually(t *testing.T, msg string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(15 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(25 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", msg) +} + +// watchDir counts filesystem events under dir, so a test can tell a quiet +// spool from one that keeps being rewritten. +func watchDir(t *testing.T, dir string) func() int { + t.Helper() + w, err := fsnotify.NewWatcher() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = w.Close() }) + if err := w.Add(dir); err != nil { + t.Fatal(err) + } + events := make(chan struct{}, 1024) + go func() { + for range w.Events { + events <- struct{}{} + } + }() + return func() int { return len(events) } +} + +func TestNatsStoreReplicasStayInSync(t *testing.T) { + cluster := newNatsCluster(t) + a, b := cluster.join(true), cluster.join(true) + ctx := context.Background() + + t.Run("config edits propagate without echoing", func(t *testing.T) { + if a.debugOn() || b.debugOn() { + t.Fatal("debug starts enabled") + } + if code := a.setDebug("true"); code != http.StatusOK { + t.Fatalf("edit status = %d", code) + } + eventually(t, "the edit to reach the other replica", b.debugOn) + + // Both spools and the bucket then stay still: nothing bounces back. + rev := a.configRevision(t) + aEvents := watchDir(t, filepath.Dir(a.store.ConfigPath())) + bEvents := watchDir(t, filepath.Dir(b.store.ConfigPath())) + time.Sleep(2 * time.Second) + if got := a.configRevision(t); got != rev { + t.Fatalf("config revision moved %d -> %d with no edit", rev, got) + } + if n := aEvents() + bEvents(); n != 0 { + t.Fatalf("spool saw %d events with no edit", n) + } + + if code := b.setDebug("false"); code != http.StatusOK { + t.Fatalf("edit status = %d", code) + } + eventually(t, "the reverse edit to reach the first replica", func() bool { return !a.debugOn() }) + }) + + t.Run("credential changes and deletes propagate", func(t *testing.T) { + const id = "team/e2e@example.com.json" + auth := &coreauth.Auth{ID: id, FileName: id, Provider: "test", Metadata: map[string]any{"type": "test", "access_token": "tok-e2e"}} + if _, err := a.store.Save(ctx, auth); err != nil { + t.Fatal(err) + } + remote := filepath.Join(b.store.AuthDir(), "team", "e2e@example.com.json") + eventually(t, "the credential to reach the other replica", func() bool { + _, err := os.Stat(remote) + return err == nil + }) + if err := a.store.Delete(ctx, id); err != nil { + t.Fatal(err) + } + eventually(t, "the delete to reach the other replica", func() bool { + _, err := os.Stat(remote) + return os.IsNotExist(err) + }) + }) + + t.Run("a stale edit is refused with 409 and adopts the winner", func(t *testing.T) { + stale := cluster.join(false) + if code := a.setDebug("true"); code != http.StatusOK { + t.Fatalf("edit status = %d", code) + } + eventually(t, "the edit to reach the synced replica", b.debugOn) + if code := stale.setDebug("true"); code != http.StatusConflict { + t.Fatalf("stale edit status = %d, want 409", code) + } + if !stale.debugOn() { + t.Fatal("the stale replica did not adopt the winning config") + } + if code := stale.setDebug("false"); code != http.StatusOK { + t.Fatalf("retry status = %d, want 200", code) + } + eventually(t, "the retry to reach the synced replicas", func() bool { return !a.debugOn() && !b.debugOn() }) + }) +}