From f9228730794e5bb45440341609a4ee911f3898a3 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Sat, 7 Nov 2026 21:01:23 -0500 Subject: [PATCH] feat(natsstore): apply other replicas' config changes to the spool A config edit on one replica now reaches the others without a restart: the watched change lands in the spool file, and the file watcher reloads it through the path every config edit already takes. The revision the next edit is checked against advances with each watched change, so an edit made after receiving one is not a false conflict, while a replica that missed one still loses its stale write. A delete of the config is never mirrored, since an empty config would wipe what every replica shares. Signed-off-by: Yordis Prieto --- internal/store/natsstore/configstore.go | 36 +++++++ internal/store/natsstore/configsync.go | 31 ++++++ internal/store/natsstore/configsync_test.go | 106 ++++++++++++++++++++ internal/store/natsstore/store.go | 19 +++- 4 files changed, 191 insertions(+), 1 deletion(-) create mode 100644 internal/store/natsstore/configsync.go create mode 100644 internal/store/natsstore/configsync_test.go diff --git a/internal/store/natsstore/configstore.go b/internal/store/natsstore/configstore.go index cff7c47a85f..932e9dee7de 100644 --- a/internal/store/natsstore/configstore.go +++ b/internal/store/natsstore/configstore.go @@ -19,6 +19,9 @@ type ConfigStore struct { mu sync.Mutex revision uint64 + // floor is the highest revision seen from any source. A watched change at + // or below it is stale or an echo of this store's own write. + floor uint64 } // NewConfigStore creates a store over the config bucket. @@ -37,6 +40,7 @@ func (s *ConfigStore) Read(ctx context.Context) ([]byte, uint64, error) { return nil, 0, err } s.revision = entry.Revision + s.noteRevision(entry.Revision) return entry.Value, entry.Revision, nil } @@ -68,9 +72,41 @@ func (s *ConfigStore) Write(ctx context.Context, data []byte) (uint64, error) { return 0, err } s.revision = rev + s.noteRevision(rev) return rev, nil } +// noteRevision raises the ordering floor. It expects s.mu to be held. +func (s *ConfigStore) noteRevision(rev uint64) { + if rev > s.floor { + s.floor = rev + } +} + +// observe takes a watched change and reports the config the spool should now +// hold, or false when there is nothing to write: the change is stale, was +// written by this replica, or deleted the config, which is never mirrored since +// an empty config would wipe what every replica shares. The revision the next +// write is checked against advances either way, so a later edit builds on the +// config that is actually current. +func (s *ConfigStore) observe(change Change) ([]byte, bool) { + s.mu.Lock() + defer s.mu.Unlock() + if change.Revision <= s.floor { + return nil, false + } + s.noteRevision(change.Revision) + if change.Deleted { + s.revision = 0 + return nil, false + } + s.revision = change.Revision + if change.Origin != "" && change.Origin == s.kv.Origin() { + return nil, false + } + return change.Value, true +} + func normalizeLineEndings(s string) string { s = strings.ReplaceAll(s, "\r\n", "\n") return strings.ReplaceAll(s, "\r", "\n") diff --git a/internal/store/natsstore/configsync.go b/internal/store/natsstore/configsync.go new file mode 100644 index 00000000000..687c0277083 --- /dev/null +++ b/internal/store/natsstore/configsync.go @@ -0,0 +1,31 @@ +package natsstore + +import ( + "bytes" + + log "github.com/sirupsen/logrus" +) + +// applyConfigEvent mirrors a watched config change into the spool, where the +// file watcher reloads it through the path every config edit already takes. +func (s *Store) applyConfigEvent(ev Event) { + if ev.Change == nil || ev.Change.ID != configID { + return + } + s.configMu.Lock() + defer s.configMu.Unlock() + data, ok := s.config.observe(*ev.Change) + if !ok { + return + } + s.mu.Lock() + same := bytes.Equal(s.configData, []byte(normalizeLineEndings(string(data)))) + s.mu.Unlock() + if same { + // Already in the spool, so rewriting it would only wake the watcher. + return + } + if err := s.writeConfigFile(data); err != nil { + log.WithError(err).Warn("nats store: failed to apply the config to the spool") + } +} diff --git a/internal/store/natsstore/configsync_test.go b/internal/store/natsstore/configsync_test.go new file mode 100644 index 00000000000..1e825559190 --- /dev/null +++ b/internal/store/natsstore/configsync_test.go @@ -0,0 +1,106 @@ +package natsstore + +import ( + "os" + "testing" + "time" +) + +// editConfig is an operator editing the spool file and the watcher persisting it. +func editConfig(t *testing.T, s *Store, content string) { + t.Helper() + if err := os.WriteFile(s.ConfigPath(), []byte(content), 0o600); err != nil { + t.Fatal(err) + } + if err := s.PersistConfig(testContext(t)); err != nil { + t.Fatal(err) + } +} + +func TestConfigSyncPropagatesEditsBothWays(t *testing.T) { + a, b, _ := syncedPair(t) + + editConfig(t, a, "port: 9001\n") + waitFor(t, "a's edit to reach b", func() bool { return fileEquals(b.ConfigPath(), "port: 9001\n") }) + + // b's edit builds on the revision its watch delivered, so it is not a conflict. + editConfig(t, b, "port: 9002\n") + waitFor(t, "b's edit to reach a", func() bool { return fileEquals(a.ConfigPath(), "port: 9002\n") }) +} + +func TestConfigSyncDoesNotEchoBackIntoTheBucket(t *testing.T) { + a, b, _ := syncedPair(t) + ctx := testContext(t) + + editConfig(t, a, "port: 9001\n") + waitFor(t, "a's edit to reach b", func() bool { return fileEquals(b.ConfigPath(), "port: 9001\n") }) + before, err := a.config.kv.Get(ctx, configID) + if err != nil { + t.Fatal(err) + } + info, err := os.Stat(a.ConfigPath()) + if err != nil { + t.Fatal(err) + } + + // What the file watcher does after reloading the config the sync wrote. + if err := b.PersistConfig(ctx); err != nil { + t.Fatal(err) + } + // And a's own write coming back through its watch. + time.Sleep(300 * time.Millisecond) + after, err := a.config.kv.Get(ctx, configID) + if err != nil || after.Revision != before.Revision { + t.Fatalf("revision moved from %d to %d (%v), want no write back", before.Revision, after.Revision, err) + } + if again, err := os.Stat(a.ConfigPath()); err != nil || !again.ModTime().Equal(info.ModTime()) { + t.Fatalf("writer's own config was rewritten (%v)", err) + } +} + +func TestConfigSyncKeepsAStaleReplicaFromOverwriting(t *testing.T) { + a, b, _ := syncedPair(t) + + b.stopSync() + editConfig(t, a, "port: 9001\n") + // b missed the change, so its edit is checked against the old revision. + if err := os.WriteFile(b.ConfigPath(), []byte("port: 9002\n"), 0o600); err != nil { + t.Fatal(err) + } + if err := b.PersistConfig(testContext(t)); err == nil { + t.Fatal("a stale replica overwrote the newer config") + } + stored, err := a.config.kv.Get(testContext(t), configID) + if err != nil || string(stored.Value) != "port: 9001\n" { + t.Fatalf("stored = %q, %v", stored.Value, err) + } +} + +func TestConfigObserveOrdersAndIgnoresDeletes(t *testing.T) { + store := NewConfigStore(NewKV("unit-config", nil, WithOrigin("me"))) + put := func(rev uint64, origin, value string) Change { + return Change{Entry: Entry{ID: configID, Value: []byte(value), Revision: rev, Origin: origin}} + } + + if data, ok := store.observe(put(4, "other", "four")); !ok || string(data) != "four" || store.Revision() != 4 { + t.Fatalf("remote put = %q, %v at %d", data, ok, store.Revision()) + } + if _, ok := store.observe(put(3, "other", "three")); ok || store.Revision() != 4 { + t.Fatalf("replayed older revision applied (revision %d)", store.Revision()) + } + if _, ok := store.observe(put(4, "other", "again")); ok { + t.Fatal("duplicate delivery applied twice") + } + // Own write: nothing for the spool, but later edits build on it. + if _, ok := store.observe(put(5, "me", "mine")); ok || store.Revision() != 5 { + t.Fatalf("own write = %v at %d", ok, store.Revision()) + } + del := put(6, "", "") + del.Deleted = true + if _, ok := store.observe(del); ok { + t.Fatal("a delete was mirrored, which would wipe the shared config") + } + if store.Revision() != 0 { + t.Fatalf("revision after delete = %d, want 0 so the next write recreates it", store.Revision()) + } +} diff --git a/internal/store/natsstore/store.go b/internal/store/natsstore/store.go index d2b89aaa84b..f6340023c2a 100644 --- a/internal/store/natsstore/store.go +++ b/internal/store/natsstore/store.go @@ -36,6 +36,11 @@ type Store struct { cancel context.CancelFunc done chan struct{} + // configMu keeps a watched config landing in the spool and the revision it + // advances to together, so a persist never reads the file between them and + // writes an older config over the newer one. + configMu sync.Mutex + mu sync.Mutex // configData is the config last written to or read from the spool, so the // watcher echoing that write back is not stored again. @@ -100,10 +105,20 @@ func (s *Store) StartSync(ctx context.Context) { ctx, s.cancel = context.WithCancel(ctx) done := make(chan struct{}) s.done = done + var wg sync.WaitGroup + wg.Add(2) go func() { - defer close(done) + defer wg.Done() s.auth.kv.Follow(ctx, s.auth.applyEvent) }() + go func() { + defer wg.Done() + s.config.kv.Follow(ctx, s.applyConfigEvent) + }() + go func() { + wg.Wait() + close(done) + }() } func (s *Store) stopSync() { @@ -252,6 +267,8 @@ func (s *Store) syncAuth(ctx context.Context) error { // file is ignored, since it is more likely a write in progress than an intent // to wipe the config every replica shares. func (s *Store) PersistConfig(ctx context.Context) error { + s.configMu.Lock() + defer s.configMu.Unlock() data, err := os.ReadFile(s.configPath) if err != nil { if errors.Is(err, fs.ErrNotExist) {