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

cmd/server/backend.go

@@ -270,6 +270,8 @@ func detectNatsBackend(env backendEnv) openBackendFunc {
 			_ = inst.Close()
 			return nil, fmt.Errorf("failed to bootstrap nats-backed config: %w", errBootstrap)
 		}
+		// The watch outlives this function's startup deadline.
+		inst.StartSync(context.Background())
 		return &openedBackend{
 			tokenStore: inst,
 			configPath: inst.ConfigPath(),

internal/store/natsstore/authstore.go

@@ -31,6 +31,9 @@ type AuthStore struct {
 	// revisions is the revision last read or written per credential, the
 	// expectation a later save is checked against.
 	revisions map[string]uint64
+	// floor is the highest revision seen per credential from any source,
+	// including deletes. A watched change at or below it is stale or an echo.
+	floor map[string]uint64
 }
 
 // NewAuthStore creates a store over the auth bucket that renders files under authDir.
@@ -42,7 +45,7 @@ func NewAuthStore(kv *KV, authDir string) (*AuthStore, error) {
 	if err := os.MkdirAll(abs, 0o700); err != nil {
 		return nil, fmt.Errorf("nats store: create auth directory: %w", err)
 	}
-	return &AuthStore{kv: kv, authDir: abs, revisions: make(map[string]uint64)}, nil
+	return &AuthStore{kv: kv, authDir: abs, revisions: make(map[string]uint64), floor: make(map[string]uint64)}, nil
 }
 
 // AuthDir returns the directory credential files are rendered into.
@@ -59,6 +62,7 @@ func (s *AuthStore) Snapshot(ctx context.Context) ([]Entry, error) {
 	s.revisions = make(map[string]uint64, len(entries))
 	for _, entry := range entries {
 		s.revisions[entry.ID] = entry.Revision
+		s.noteRevision(entry.ID, entry.Revision)
 	}
 	return entries, nil
 }
@@ -170,6 +174,7 @@ func (s *AuthStore) Save(ctx context.Context, auth *cliproxyauth.Auth) (string,
 		}
 		if err == nil {
 			s.revisions[id] = rev
+			s.noteRevision(id, rev)
 			s.finishSave(auth, path)
 			return path, nil
 		}
@@ -189,6 +194,7 @@ func (s *AuthStore) Save(ctx context.Context, auth *cliproxyauth.Auth) (string,
 			return "", err
 		default:
 			s.revisions[id] = current.Revision
+			s.noteRevision(id, current.Revision)
 			if err := s.adoptRemote(path, current.Value, auth); err != nil {
 				return "", err
 			}
@@ -420,11 +426,13 @@ func (s *AuthStore) PutFile(ctx context.Context, id string, data []byte) error {
 			return err
 		}
 		s.revisions[id] = rev
+		s.noteRevision(id, rev)
 		return nil
 	case err != nil:
 		return err
 	case sameJSON(current.Value, data):
 		s.revisions[id] = current.Revision
+		s.noteRevision(id, current.Revision)
 		return nil
 	case current.Revision != expected:
 		return &ConflictError{Bucket: s.kv.bucket, ID: id}
@@ -434,5 +442,34 @@ func (s *AuthStore) PutFile(ctx context.Context, id string, data []byte) error {
 		return err
 	}
 	s.revisions[id] = rev
+	s.noteRevision(id, rev)
+	return nil
+}
+
+// noteRevision raises the ordering floor. It expects s.mu to be held.
+func (s *AuthStore) noteRevision(id string, rev uint64) {
+	if rev > s.floor[id] {
+		s.floor[id] = rev
+	}
+}
+
+// DeleteObserved removes a credential whose file disappeared from the spool,
+// but only the revision this store has seen. A credential it never saw, or one
+// that changed since, is left alone: the file went missing because a remote
+// delete was applied or a remote change has not arrived yet, not because the
+// credential should go.
+func (s *AuthStore) DeleteObserved(ctx context.Context, id string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	observed := s.revisions[id]
+	if observed == 0 {
+		return nil
+	}
+	err := s.kv.Delete(ctx, id, observed)
+	if err != nil && !errors.Is(err, ErrConflict) {
+		return err
+	}
+	// Either the delete landed or a newer revision won; the watch reports it.
+	delete(s.revisions, id)
 	return nil
 }

internal/store/natsstore/authsync.go

@@ -0,0 +1,98 @@
+package natsstore
+
+import (
+	"bytes"
+	"errors"
+	"io/fs"
+	"os"
+	"path/filepath"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// applyEvent mirrors a watched bucket event into the spool, where the file
+// watcher and synthesizer pick it up like any other edit.
+func (s *AuthStore) applyEvent(ev Event) {
+	if ev.Synced {
+		s.prune(ev.Initial, ev.InitialMax)
+		return
+	}
+	if ev.Change == nil {
+		return
+	}
+	if err := s.applyChange(*ev.Change); err != nil {
+		log.WithError(err).Warnf("nats store: failed to apply auth %s to the spool", ev.Change.ID)
+	}
+}
+
+// applyChange writes or removes one credential file. A change at or below the
+// highest revision already seen is dropped, which orders replays and discards
+// the bucket echoing this store's own writes.
+func (s *AuthStore) applyChange(change Change) error {
+	path, err := s.absoluteAuthPath(change.ID)
+	if err != nil {
+		return err
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if change.Revision <= s.floor[change.ID] {
+		return nil
+	}
+	if change.Deleted {
+		s.noteRevision(change.ID, change.Revision)
+		delete(s.revisions, change.ID)
+		return removeSpoolFile(path)
+	}
+	// The spool already holds what this replica wrote, and may hold a newer
+	// local edit that rewriting it would lose.
+	own := change.Origin != "" && change.Origin == s.kv.Origin()
+	if !own {
+		if err := writeSpoolFile(path, change.Value); err != nil {
+			return err
+		}
+	}
+	s.noteRevision(change.ID, change.Revision)
+	s.revisions[change.ID] = change.Revision
+	return nil
+}
+
+// prune forgets credentials the bucket no longer holds. A delete that happened
+// while the watch was down leaves no event behind once its marker is gone, so
+// the replay is compared with what this store last saw. A credential written
+// after the replay was taken is newer than every replayed revision and stays.
+func (s *AuthStore) prune(initial map[string]struct{}, initialMax uint64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for id, rev := range s.revisions {
+		if _, ok := initial[id]; ok || rev > initialMax {
+			continue
+		}
+		path, err := s.absoluteAuthPath(id)
+		if err != nil {
+			continue
+		}
+		delete(s.revisions, id)
+		if err := removeSpoolFile(path); err != nil {
+			log.WithError(err).Warnf("nats store: failed to remove auth %s from the spool", id)
+		}
+	}
+}
+
+// writeSpoolFile replaces a spool file unless it already holds the bytes, so a
+// repeated value produces no file event for the watcher to react to.
+func writeSpoolFile(path string, data []byte) error {
+	if current, err := os.ReadFile(path); err == nil && bytes.Equal(current, data) {
+		return nil
+	}
+	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
+		return err
+	}
+	return writeFileAtomic(path, data)
+}
+
+func removeSpoolFile(path string) error {
+	if err := os.Remove(path); err != nil && !errors.Is(err, fs.ErrNotExist) {
+		return err
+	}
+	return nil
+}