feat(natsstore): apply other replicas' config changes to the spool

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

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

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

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