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) {