Change Fetch Mechanism

src/cluster/mod.rs

@@ -363,13 +363,22 @@ fn spawn_client_build(cx: &mut App, context: ContextInfo) {
                 Ok(Err(error)) => {
                     tracing::warn!(%error, context = %context_id, "cluster: client build failed");
                     cx.update_global::<ClusterState, _>(|state, _| {
+                        // The active label now points at a context with no
+                        // working client. Drop the old one and let every view
+                        // resubscribe into the visible error state.
+                        state.client = None;
                         state.reachable = Some(false);
+                        state.namespaces.clear();
+                        state.context_epoch = state.context_epoch.wrapping_add(1);
                     });
                 }
                 Err(error) => {
                     tracing::error!(%error, context = %context_id, "cluster: client build task failed");
                     cx.update_global::<ClusterState, _>(|state, _| {
+                        state.client = None;
                         state.reachable = Some(false);
+                        state.namespaces.clear();
+                        state.context_epoch = state.context_epoch.wrapping_add(1);
                     });
                 }
             }
@@ -390,11 +399,11 @@ pub fn switch_context(cx: &mut App, id: &str) -> Result<(), &'static str> {
             .find(|context| context.id == id)
             .cloned()
             .ok_or("unknown context")?;
-        state.client = None;
+        // Keep the old client and snapshots until the new client is ready.
+        // Bumping the epoch here would make every view resubscribe with no
+        // client, publish an error, and resubscribe again on success.
         state.active = Some(context.id.clone());
         state.reachable = None;
-        state.namespaces.clear();
-        state.context_epoch = state.context_epoch.wrapping_add(1);
         if let Some(cancel) = state.namespace_cancel.take() {
             let _ = cancel.send(true);
         }

src/models/resources/store.rs

@@ -5,7 +5,7 @@
 //! without a separate sort pass. A UID index lets deletes/updates find the prior
 //! key in O(1) even if metadata changes.
 
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::sync::Arc;
 
 use super::Item;
@@ -33,7 +33,6 @@ impl ObjectKey {
 #[derive(Clone, Debug)]
 pub struct ResourceSnapshot {
     items: Arc<[Arc<Item>]>,
-    revision: u64,
     phase: SnapshotPhase,
     mode: WatchMode,
 }
@@ -79,7 +78,6 @@ impl ResourceSnapshot {
     pub fn loading() -> Self {
         Self {
             items: Arc::from(Vec::<Arc<Item>>::new()),
-            revision: 0,
             phase: SnapshotPhase::Loading,
             mode: WatchMode::Watch,
         }
@@ -88,7 +86,6 @@ impl ResourceSnapshot {
     pub fn failed(message: impl Into<String>) -> Self {
         Self {
             items: Arc::from(Vec::<Arc<Item>>::new()),
-            revision: 0,
             phase: SnapshotPhase::Error(message.into()),
             mode: WatchMode::Watch,
         }
@@ -103,10 +100,6 @@ impl ResourceSnapshot {
         &self.items
     }
 
-    pub fn revision(&self) -> u64 {
-        self.revision
-    }
-
     pub fn phase(&self) -> &SnapshotPhase {
         &self.phase
     }
@@ -120,8 +113,10 @@ impl ResourceSnapshot {
 pub struct ResourceStore {
     objects: BTreeMap<ObjectKey, Arc<Item>>,
     by_uid: HashMap<String, ObjectKey>,
-    revision: u64,
     pending: usize,
+    /// Keys seen since [`begin_sync`](Self::begin_sync). Present only while a
+    /// list is being merged into the store.
+    sync_seen: Option<HashSet<ObjectKey>>,
 }
 
 impl Default for ResourceStore {
@@ -135,8 +130,8 @@ impl ResourceStore {
         Self {
             objects: BTreeMap::new(),
             by_uid: HashMap::new(),
-            revision: 0,
             pending: 0,
+            sync_seen: None,
         }
     }
 
@@ -148,47 +143,122 @@ impl ResourceStore {
         self.pending
     }
 
+    #[cfg(test)]
     pub fn reset<I>(&mut self, items: I)
     where
         I: IntoIterator<Item = Item>,
     {
         self.objects.clear();
         self.by_uid.clear();
+        self.sync_seen = None;
         for item in items {
             self.upsert(item);
         }
         self.pending = 0;
     }
 
-    pub fn upsert(&mut self, item: Item) {
-        let item = Arc::new(item);
+    /// Start merging a full list into the store. The old objects stay visible
+    /// until [`finish_sync`](Self::finish_sync), so a relist does not blank the
+    /// UI while its pages arrive.
+    pub fn begin_sync(&mut self) {
+        self.sync_seen = Some(HashSet::new());
+    }
+
+    /// Remove objects that were absent from the list started by
+    /// [`begin_sync`](Self::begin_sync). Returns whether anything was removed.
+    pub fn finish_sync(&mut self) -> bool {
+        let Some(seen) = self.sync_seen.take() else {
+            return false;
+        };
+        let stale: Vec<ObjectKey> = self
+            .objects
+            .keys()
+            .filter(|key| !seen.contains(*key))
+            .cloned()
+            .collect();
+        let mut changed = false;
+        for key in stale {
+            changed |= self.remove(&key);
+        }
+        changed
+    }
+
+    /// Abandon an interrupted list without deleting objects it did not reach.
+    pub fn abort_sync(&mut self) {
+        self.sync_seen = None;
+    }
+
+    /// Insert or replace an object. Returns whether the store actually changed:
+    /// a repeated `(key, uid, resourceVersion)` is a no-op, which makes poll
+    /// relists and duplicate watch events cheap.
+    pub fn upsert(&mut self, item: Item) -> bool {
         let key = ObjectKey::from_item(&item);
+        if let Some(seen) = &mut self.sync_seen {
+            seen.insert(key.clone());
+        }
+
+        if let Some(existing) = self.objects.get(&key)
+            && !item.resource_version.is_empty()
+            && existing.uid == item.uid
+            && existing.resource_version == item.resource_version
+        {
+            return false;
+        }
+
+        let item = Arc::new(item);
+        if let Some(existing) = self.objects.get(&key)
+            && !existing.uid.is_empty()
+            && existing.uid != item.uid
+        {
+            self.by_uid.remove(&existing.uid);
+        }
         if !item.uid.is_empty()
             && let Some(previous) = self.by_uid.insert(item.uid.clone(), key.clone())
             && previous != key
         {
             self.objects.remove(&previous);
         }
         self.objects.insert(key, item);
-        self.revision = self.revision.wrapping_add(1);
         self.pending = self.pending.saturating_add(1);
+        true
     }
 
-    pub fn remove(&mut self, key: &ObjectKey) {
-        if let Some(item) = self.objects.remove(key)
-            && !item.uid.is_empty()
-        {
+    pub fn remove(&mut self, key: &ObjectKey) -> bool {
+        let Some(item) = self.objects.remove(key) else {
+            return false;
+        };
+        if !item.uid.is_empty() {
             self.by_uid.remove(&item.uid);
         }
-        self.revision = self.revision.wrapping_add(1);
         self.pending = self.pending.saturating_add(1);
+        true
+    }
+
+    /// Remove the oldest objects until at most `max_items` remain. Used for
+    /// high-churn kinds such as Events, where the API can retain thousands.
+    /// Returns whether anything was removed.
+    pub fn trim_oldest(&mut self, max_items: usize) -> bool {
+        if self.objects.len() <= max_items {
+            return false;
+        }
+        let mut by_age: Vec<(ObjectKey, i64)> = self
+            .objects
+            .iter()
+            .map(|(key, item)| (key.clone(), item.created_at.unwrap_or(i64::MIN)))
+            .collect();
+        by_age.sort_by_key(|(_, created_at)| *created_at);
+        let remove = self.objects.len() - max_items;
+        let mut changed = false;
+        for (key, _) in by_age.into_iter().take(remove) {
+            changed |= self.remove(&key);
+        }
+        changed
     }
 
     pub fn snapshot(&mut self, phase: SnapshotPhase) -> ResourceSnapshot {
         self.pending = 0;
         ResourceSnapshot {
             items: self.objects.values().cloned().collect(),
-            revision: self.revision,
             phase,
             mode: WatchMode::Watch,
         }
@@ -205,6 +275,12 @@ mod tests {
         item
     }
 
+    fn versioned_pod(name: &str, namespace: &str, uid: &str, version: &str) -> Item {
+        let mut item = pod(name, namespace, uid);
+        item.resource_version = version.into();
+        item
+    }
+
     #[test]
     fn keeps_rows_sorted_and_replaces_by_key() {
         let mut store = ResourceStore::new();
@@ -220,7 +296,6 @@ mod tests {
         let names: Vec<_> = snapshot.items().iter().map(|i| i.name.as_str()).collect();
         assert_eq!(names, ["a", "b", "c"]);
         assert_eq!(snapshot.items()[1].status, "crashloopbackoff");
-        assert_eq!(snapshot.revision(), 4);
     }
 
     #[test]
@@ -273,6 +348,103 @@ mod tests {
         assert_eq!(snapshot.items()[0].name, "b");
     }
 
+    #[test]
+    fn upsert_skips_an_unchanged_resource_version() {
+        let mut store = ResourceStore::new();
+        store.upsert(versioned_pod("a", "ns", "uid-a", "10"));
+        let _ = store.snapshot(SnapshotPhase::Ready);
+
+        let mut same = versioned_pod("a", "ns", "uid-a", "10");
+        same.status = "running".into();
+        assert!(!store.upsert(same));
+        assert_eq!(store.pending(), 0);
+        assert_eq!(store.snapshot(SnapshotPhase::Ready).items()[0].status, "");
+    }
+
+    #[test]
+    fn sync_keeps_unchanged_items_and_removes_absent_ones() {
+        let mut store = ResourceStore::new();
+        store.reset([
+            versioned_pod("a", "ns", "uid-a", "1"),
+            versioned_pod("b", "ns", "uid-b", "1"),
+            versioned_pod("c", "ns", "uid-c", "1"),
+        ]);
+        let _ = store.snapshot(SnapshotPhase::Ready);
+
+        store.begin_sync();
+        assert!(!store.upsert(versioned_pod("a", "ns", "uid-a", "1")));
+        assert!(store.upsert(versioned_pod("b", "ns", "uid-b", "2")));
+        assert!(store.upsert(versioned_pod("d", "ns", "uid-d", "1")));
+        store.finish_sync();
+
+        let snapshot = store.snapshot(SnapshotPhase::Ready);
+        let names: Vec<_> = snapshot.items().iter().map(|i| i.name.as_str()).collect();
+        assert_eq!(names, ["a", "b", "d"]);
+    }
+
+    #[test]
+    fn an_unchanged_sync_does_not_bump_pending() {
+        let mut store = ResourceStore::new();
+        store.reset([versioned_pod("a", "ns", "uid-a", "1")]);
+        let _ = store.snapshot(SnapshotPhase::Ready);
+
+        store.begin_sync();
+        assert!(!store.upsert(versioned_pod("a", "ns", "uid-a", "1")));
+        store.finish_sync();
+
+        assert_eq!(store.pending(), 0);
+        assert_eq!(store.snapshot(SnapshotPhase::Ready).items().len(), 1);
+    }
+
+    #[test]
+    fn aborting_a_sync_does_not_delete_unseen_items() {
+        let mut store = ResourceStore::new();
+        store.reset([
+            versioned_pod("a", "ns", "uid-a", "1"),
+            versioned_pod("b", "ns", "uid-b", "1"),
+        ]);
+        let _ = store.snapshot(SnapshotPhase::Ready);
+
+        store.begin_sync();
+        store.upsert(versioned_pod("a", "ns", "uid-a", "2"));
+        store.abort_sync();
+
+        let snapshot = store.snapshot(SnapshotPhase::Ready);
+        assert_eq!(snapshot.items().len(), 2);
+    }
+
+    #[test]
+    fn removing_an_unknown_key_is_a_no_op() {
+        let mut store = ResourceStore::new();
+        let _ = store.snapshot(SnapshotPhase::Ready);
+        assert!(!store.remove(&ObjectKey::new("ns", "missing")));
+        assert_eq!(store.pending(), 0);
+    }
+
+    #[test]
+    fn trim_oldest_removes_the_oldest_objects_first() {
+        let mut store = ResourceStore::new();
+        for (index, created_at) in [(0, 10), (1, 30), (2, 20), (3, 40)] {
+            let mut item = versioned_pod(
+                &format!("event-{index}"),
+                "ns",
+                &format!("uid-{index}"),
+                "1",
+            );
+            item.created_at = Some(created_at);
+            store.upsert(item);
+        }
+        let _ = store.snapshot(SnapshotPhase::Ready);
+
+        assert!(store.trim_oldest(2));
+        let snapshot = store.snapshot(SnapshotPhase::Ready);
+        let names: Vec<_> = snapshot.items().iter().map(|i| i.name.as_str()).collect();
+        assert_eq!(names, ["event-1", "event-3"]);
+
+        assert!(!store.trim_oldest(2));
+        assert_eq!(store.pending(), 0);
+    }
+
     #[test]
     fn snapshots_carry_the_watch_mode() {
         let mut store = ResourceStore::new();