From 6511368319d375e8947d3c2eeefb95e340471de8 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Sat, 7 Nov 2026 20:58:19 -0500 Subject: [PATCH] feat(natsstore): apply other replicas' credential changes to the spool Replicas only saw each other's credentials after a restart. Following the auth bucket and writing its changes into the spool lets the file watcher and synthesizer pick them up unchanged. A change at or below the highest revision already seen is dropped, which orders replays and discards the bucket echoing this replica's own writes without overwriting a newer local edit. An applied file equal to the spool makes no write, and removing a file only deletes the revision this replica saw, so applying a change never feeds back into the bucket. Signed-off-by: Yordis Prieto --- cmd/server/backend.go | 2 + internal/store/natsstore/authstore.go | 39 ++- internal/store/natsstore/authsync.go | 98 +++++++ internal/store/natsstore/authsync_test.go | 309 ++++++++++++++++++++++ internal/store/natsstore/kv.go | 2 + internal/store/natsstore/store.go | 44 ++- internal/store/natsstore/watch.go | 92 +++++++ 7 files changed, 582 insertions(+), 4 deletions(-) create mode 100644 internal/store/natsstore/authsync.go create mode 100644 internal/store/natsstore/authsync_test.go create mode 100644 internal/store/natsstore/watch.go diff --git a/cmd/server/backend.go b/cmd/server/backend.go index ee12e853eec..245a683b956 100644 --- a/cmd/server/backend.go +++ b/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(), diff --git a/internal/store/natsstore/authstore.go b/internal/store/natsstore/authstore.go index 28b1fe7e440..55028fa6acd 100644 --- a/internal/store/natsstore/authstore.go +++ b/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 } diff --git a/internal/store/natsstore/authsync.go b/internal/store/natsstore/authsync.go new file mode 100644 index 00000000000..09f0237f22f --- /dev/null +++ b/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 +} diff --git a/internal/store/natsstore/authsync_test.go b/internal/store/natsstore/authsync_test.go new file mode 100644 index 00000000000..92e6505da6a --- /dev/null +++ b/internal/store/natsstore/authsync_test.go @@ -0,0 +1,309 @@ +package natsstore + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +// waitFor polls until cond holds, failing the test with msg on timeout. +func waitFor(t *testing.T, msg string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", msg) +} + +func fileEquals(path, want string) bool { + got, err := os.ReadFile(path) + return err == nil && string(got) == want +} + +func fileMissing(path string) bool { + _, err := os.Stat(path) + return os.IsNotExist(err) +} + +// syncedPair returns two bootstrapped replicas that follow the buckets. +func syncedPair(t *testing.T) (a, b *Store, f *storeFixture) { + t.Helper() + f = newStoreFixture(t) + a, b = f.open(), f.open() + for _, s := range []*Store{a, b} { + if err := s.Bootstrap(testContext(t), f.example); err != nil { + t.Fatal(err) + } + s.StartSync(context.Background()) + t.Cleanup(func() { _ = s.Close() }) + } + return a, b, f +} + +func TestAuthSyncPropagatesSavesAndDeletes(t *testing.T) { + a, b, _ := syncedPair(t) + ctx := testContext(t) + const id = "team/user@example.com.json" + remote := filepath.Join(b.AuthDir(), "team", "user@example.com.json") + local := filepath.Join(a.AuthDir(), "team", "user@example.com.json") + + if _, err := a.Save(ctx, testAuth(id, map[string]any{"type": "test", "access_token": "tok-1"})); err != nil { + t.Fatal(err) + } + want := readFile(t, local) + waitFor(t, "create to reach the other spool", func() bool { return fileEquals(remote, want) }) + + if _, err := a.Save(ctx, testAuth(id, map[string]any{"type": "test", "access_token": "tok-2"})); err != nil { + t.Fatal(err) + } + want = readFile(t, local) + waitFor(t, "update to reach the other spool", func() bool { return fileEquals(remote, want) }) + + if err := a.Delete(ctx, id); err != nil { + t.Fatal(err) + } + waitFor(t, "delete to reach the other spool", func() bool { return fileMissing(remote) }) +} + +func TestAuthSyncDoesNotEchoBackIntoTheBucket(t *testing.T) { + a, b, _ := syncedPair(t) + ctx := testContext(t) + const id = "echo.json" + remote := filepath.Join(b.AuthDir(), id) + + if _, err := a.Save(ctx, testAuth(id, map[string]any{"type": "test", "access_token": "tok"})); err != nil { + t.Fatal(err) + } + waitFor(t, "create to reach the other spool", func() bool { _, err := os.Stat(remote); return err == nil }) + before, err := a.auth.kv.Get(ctx, id) + if err != nil { + t.Fatal(err) + } + + // What the file watcher does when it sees the spool file the sync wrote. + if err := b.PersistAuthFiles(ctx, "echo", remote); err != nil { + t.Fatal(err) + } + // The writer's own change comes back through its watch and must not rewrite the file. + info, err := os.Stat(filepath.Join(a.AuthDir(), id)) + if err != nil { + t.Fatal(err) + } + time.Sleep(300 * time.Millisecond) + after, err := a.auth.kv.Get(ctx, id) + 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) + } + again, err := os.Stat(filepath.Join(a.AuthDir(), id)) + if err != nil || !again.ModTime().Equal(info.ModTime()) { + t.Fatalf("writer's own file was rewritten (%v)", err) + } + + // A remote delete removes the file; persisting that removal must not delete again. + if err := a.Delete(ctx, id); err != nil { + t.Fatal(err) + } + waitFor(t, "delete to reach the other spool", func() bool { return fileMissing(remote) }) + // The credential comes back while b has not applied it yet, and the + // persister handles the removal of the file the earlier delete left + // behind: the new credential must survive. + b.stopSync() + if _, err := a.Save(ctx, testAuth(id, map[string]any{"type": "test", "access_token": "tok-new"})); err != nil { + t.Fatal(err) + } + if err := b.PersistAuthFiles(ctx, "stale removal", remote); err != nil { + t.Fatal(err) + } + if _, err := a.auth.kv.Get(ctx, id); err != nil { + t.Fatalf("recreated credential was deleted: %v", err) + } +} + +func TestAuthSyncCatchesUpOnStart(t *testing.T) { + f := newStoreFixture(t) + ctx := testContext(t) + a, b := f.open(), f.open() + for _, s := range []*Store{a, b} { + if err := s.Bootstrap(ctx, f.example); err != nil { + t.Fatal(err) + } + } + // Changes between b's bootstrap and its watch starting must still arrive. + if _, err := a.Save(ctx, testAuth("late.json", map[string]any{"type": "test", "access_token": "tok"})); err != nil { + t.Fatal(err) + } + b.StartSync(context.Background()) + t.Cleanup(func() { _ = b.Close() }) + want := readFile(t, filepath.Join(a.AuthDir(), "late.json")) + waitFor(t, "catch-up", func() bool { return fileEquals(filepath.Join(b.AuthDir(), "late.json"), want) }) +} + +func TestAuthSyncResumesAfterTheWatchDrops(t *testing.T) { + old := watchRetryDelay + watchRetryDelay = 50 * time.Millisecond + t.Cleanup(func() { watchRetryDelay = old }) + + f := newStoreFixture(t) + ctx := testContext(t) + a, b := f.open(), f.open() + watches := make(chan jetstream.KeyWatcher, 8) + b.auth.kv.onWatch = func(w jetstream.KeyWatcher) { watches <- w } + for _, s := range []*Store{a, b} { + if err := s.Bootstrap(ctx, f.example); err != nil { + t.Fatal(err) + } + } + for _, id := range []string{"keep.json", "gone.json"} { + if _, err := a.Save(ctx, testAuth(id, map[string]any{"type": "test", "access_token": id})); err != nil { + t.Fatal(err) + } + } + b.StartSync(context.Background()) + t.Cleanup(func() { _ = b.Close() }) + first := <-watches + waitFor(t, "initial replay", func() bool { + return !fileMissing(filepath.Join(b.AuthDir(), "keep.json")) && !fileMissing(filepath.Join(b.AuthDir(), "gone.json")) + }) + + // While b's watch is down: a change, a delete and a new credential. + if err := first.Stop(); err != nil { + t.Fatal(err) + } + if _, err := a.Save(ctx, testAuth("keep.json", map[string]any{"type": "test", "access_token": "changed"})); err != nil { + t.Fatal(err) + } + if err := a.Delete(ctx, "gone.json"); err != nil { + t.Fatal(err) + } + if _, err := a.Save(ctx, testAuth("fresh.json", map[string]any{"type": "test", "access_token": "fresh"})); err != nil { + t.Fatal(err) + } + + select { + case <-watches: + case <-time.After(10 * time.Second): + t.Fatal("the watch was not reopened") + } + for _, id := range []string{"keep.json", "fresh.json"} { + want := readFile(t, filepath.Join(a.AuthDir(), id)) + waitFor(t, id+" to converge", func() bool { return fileEquals(filepath.Join(b.AuthDir(), id), want) }) + } + waitFor(t, "delete made while down", func() bool { return fileMissing(filepath.Join(b.AuthDir(), "gone.json")) }) +} + +// unitAuthStore has no bucket; it only exercises how watched changes reach the spool. +func unitAuthStore(t *testing.T) *AuthStore { + t.Helper() + store, err := NewAuthStore(NewKV("unit-auth", nil, WithOrigin("me")), t.TempDir()) + if err != nil { + t.Fatal(err) + } + return store +} + +func change(id string, rev uint64, origin, value string) Change { + return Change{Entry: Entry{ID: id, Value: []byte(value), Revision: rev, Origin: origin}} +} + +func TestApplyChangeOrdersByRevision(t *testing.T) { + store := unitAuthStore(t) + path := filepath.Join(store.AuthDir(), "a.json") + + for _, c := range []Change{ + change("a.json", 5, "other", "five"), + change("a.json", 3, "other", "three"), // replayed older revision + change("a.json", 5, "other", "again"), // duplicate delivery + } { + if err := store.applyChange(c); err != nil { + t.Fatal(err) + } + } + if got := readFile(t, path); got != "five" { + t.Fatalf("file = %q, want the highest revision only", got) + } + + // A delete is ordered too, and a stale put cannot resurrect the file. + del := change("a.json", 7, "", "") + del.Deleted = true + if err := store.applyChange(del); err != nil { + t.Fatal(err) + } + if err := store.applyChange(change("a.json", 6, "other", "six")); err != nil { + t.Fatal(err) + } + if !fileMissing(path) { + t.Fatal("a put older than the delete brought the file back") + } + if store.revisions["a.json"] != 0 { + t.Fatalf("revision kept after delete: %d", store.revisions["a.json"]) + } +} + +func TestApplyChangeLeavesOwnWritesAndIdenticalFilesAlone(t *testing.T) { + store := unitAuthStore(t) + path := filepath.Join(store.AuthDir(), "a.json") + if err := os.WriteFile(path, []byte("local edit"), 0o600); err != nil { + t.Fatal(err) + } + + // The bucket echoing this replica's write must not replace a newer local edit. + if err := store.applyChange(change("a.json", 2, "me", "stored")); err != nil { + t.Fatal(err) + } + if got := readFile(t, path); got != "local edit" { + t.Fatalf("own write rewrote the spool: %q", got) + } + if store.revisions["a.json"] != 2 { + t.Fatalf("revision = %d, want it advanced so the next save checks against it", store.revisions["a.json"]) + } + + // A remote value equal to the file produces no write, so no file event. + if err := os.WriteFile(path, []byte("same"), 0o600); err != nil { + t.Fatal(err) + } + past := time.Now().Add(-time.Hour) + if err := os.Chtimes(path, past, past); err != nil { + t.Fatal(err) + } + if err := store.applyChange(change("a.json", 3, "other", "same")); err != nil { + t.Fatal(err) + } + if info, err := os.Stat(path); err != nil || info.ModTime().After(past.Add(time.Minute)) { + t.Fatalf("identical content was rewritten (%v)", err) + } +} + +func TestApplyChangeRefusesPathsOutsideTheSpool(t *testing.T) { + store := unitAuthStore(t) + if err := store.applyChange(change("../escape.json", 1, "other", "x")); err == nil { + t.Fatal("a credential id escaping the spool was applied") + } +} + +func TestPruneForgetsOnlyCredentialsOlderThanTheReplay(t *testing.T) { + store := unitAuthStore(t) + for id, rev := range map[string]uint64{"live.json": 4, "gone.json": 3, "newer.json": 9} { + if err := store.applyChange(change(id, rev, "other", id)); err != nil { + t.Fatal(err) + } + } + store.prune(map[string]struct{}{"live.json": {}}, 5) + + if fileMissing(filepath.Join(store.AuthDir(), "live.json")) { + t.Fatal("a credential in the replay was pruned") + } + if !fileMissing(filepath.Join(store.AuthDir(), "gone.json")) { + t.Fatal("a credential absent from the replay was kept") + } + if fileMissing(filepath.Join(store.AuthDir(), "newer.json")) { + t.Fatal("a credential written after the replay was pruned") + } +} diff --git a/internal/store/natsstore/kv.go b/internal/store/natsstore/kv.go index bddbbfbd89e..a0d81214e5b 100644 --- a/internal/store/natsstore/kv.go +++ b/internal/store/natsstore/kv.go @@ -50,6 +50,8 @@ type KV struct { bucket string kv jetstream.KeyValue origin string + // onWatch observes each watch Follow opens, for tests that drop it. + onWatch func(jetstream.KeyWatcher) } // KVOption customizes a KV. diff --git a/internal/store/natsstore/store.go b/internal/store/natsstore/store.go index b487691bd67..d2b89aaa84b 100644 --- a/internal/store/natsstore/store.go +++ b/internal/store/natsstore/store.go @@ -32,6 +32,10 @@ type Store struct { spoolRoot string configPath string + syncMu sync.Mutex + cancel context.CancelFunc + done chan struct{} + 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. @@ -83,9 +87,43 @@ func newReplicaID() (string, error) { // ReplicaID identifies this process in the values it writes. func (s *Store) ReplicaID() string { return s.replicaID } -// Close closes the NATS connection. +// StartSync follows the buckets until the store is closed or ctx ends, mirroring +// changes other replicas make into the spool. It returns at once, runs in the +// background and does nothing when already started. Bootstrap first, so the +// watch only has to catch up on what changed since. +func (s *Store) StartSync(ctx context.Context) { + s.syncMu.Lock() + defer s.syncMu.Unlock() + if s.cancel != nil { + return + } + ctx, s.cancel = context.WithCancel(ctx) + done := make(chan struct{}) + s.done = done + go func() { + defer close(done) + s.auth.kv.Follow(ctx, s.auth.applyEvent) + }() +} + +func (s *Store) stopSync() { + s.syncMu.Lock() + cancel, done := s.cancel, s.done + s.cancel, s.done = nil, nil + s.syncMu.Unlock() + if cancel != nil { + cancel() + <-done + } +} + +// Close stops following the buckets and closes the NATS connection. func (s *Store) Close() error { - if s != nil && s.nc != nil { + if s == nil { + return nil + } + s.stopSync() + if s.nc != nil { s.nc.Close() } return nil @@ -257,7 +295,7 @@ func (s *Store) PersistAuthFiles(ctx context.Context, _ string, paths ...string) return fmt.Errorf("nats store: read auth file: %w", err) } if err != nil || len(data) == 0 { - if err := s.auth.Delete(ctx, id); err != nil { + if err := s.auth.DeleteObserved(ctx, id); err != nil { return err } continue diff --git a/internal/store/natsstore/watch.go b/internal/store/natsstore/watch.go new file mode 100644 index 00000000000..28fb98fd51e --- /dev/null +++ b/internal/store/natsstore/watch.go @@ -0,0 +1,92 @@ +package natsstore + +import ( + "context" + "time" + + "github.com/nats-io/nats.go/jetstream" + log "github.com/sirupsen/logrus" +) + +// watchRetryDelay is how long Follow waits before reopening a watch that ended. +var watchRetryDelay = time.Second + +// Change is one revision observed on a key. +type Change struct { + Entry + // Deleted is set for a delete or purge marker, whose Value is empty. + Deleted bool +} + +// Event is what Follow hands to its callback. +type Event struct { + // Change is the observed revision, nil for the synced event. + Change *Change + // Synced marks the end of the initial replay: every key that was live when + // the watch opened has been delivered. Updates after it are live changes. + Synced bool + // Initial lists the keys delivered by the replay and InitialMax is the + // highest revision among them, so the caller can tell a key the bucket no + // longer holds from one written after the replay. Both are set on Synced. + Initial map[string]struct{} + InitialMax uint64 +} + +// Follow replays the bucket and then streams its changes to apply until ctx +// ends. A watch that ends, or fails to open, is reopened after a delay and +// replays from the start, which is safe because every revision carries its +// own ordering. apply is called from one goroutine at a time. +func (k *KV) Follow(ctx context.Context, apply func(Event)) { + for ctx.Err() == nil { + watcher, err := k.kv.WatchAll(ctx) + if err != nil { + log.WithError(err).Warnf("nats store: watching %s failed, retrying", k.bucket) + } else { + if k.onWatch != nil { + k.onWatch(watcher) + } + k.consume(ctx, watcher, apply) + _ = watcher.Stop() + } + select { + case <-ctx.Done(): + case <-time.After(watchRetryDelay): + } + } +} + +func (k *KV) consume(ctx context.Context, watcher jetstream.KeyWatcher, apply func(Event)) { + var ( + synced bool + initial = make(map[string]struct{}) + initialMax uint64 + ) + for { + select { + case <-ctx.Done(): + return + case entry, ok := <-watcher.Updates(): + if !ok { + return + } + if entry == nil { + synced = true + apply(Event{Synced: true, Initial: initial, InitialMax: initialMax}) + continue + } + id, err := DecodeKey(entry.Key()) + if err != nil { + // Not written by this backend; leave it alone. + continue + } + change := Change{Entry: entryOf(id, entry), Deleted: entry.Operation() != jetstream.KeyValuePut} + if !synced { + initial[id] = struct{}{} + if change.Revision > initialMax { + initialMax = change.Revision + } + } + apply(Event{Change: &change}) + } + } +}