diff --git a/src/cluster/mod.rs b/src/cluster/mod.rs index 2bf16da..29a9a7b 100644 --- a/src/cluster/mod.rs +++ b/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::(|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::(|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); } diff --git a/src/models/resources/store.rs b/src/models/resources/store.rs index ecdda16..4f67ba6 100644 --- a/src/models/resources/store.rs +++ b/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]>, - revision: u64, phase: SnapshotPhase, mode: WatchMode, } @@ -79,7 +78,6 @@ impl ResourceSnapshot { pub fn loading() -> Self { Self { items: Arc::from(Vec::>::new()), - revision: 0, phase: SnapshotPhase::Loading, mode: WatchMode::Watch, } @@ -88,7 +86,6 @@ impl ResourceSnapshot { pub fn failed(message: impl Into) -> Self { Self { items: Arc::from(Vec::>::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>, by_uid: HashMap, - 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>, } 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,21 +143,75 @@ impl ResourceStore { self.pending } + #[cfg(test)] pub fn reset(&mut self, items: I) where I: IntoIterator, { 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 = 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 @@ -170,25 +219,46 @@ impl ResourceStore { 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(); diff --git a/src/repository/mod.rs b/src/repository/mod.rs index 1b9a79f..9d964e4 100644 --- a/src/repository/mod.rs +++ b/src/repository/mod.rs @@ -50,6 +50,8 @@ pub fn init(cx: &mut App) { #[derive(Clone)] pub struct Subscription { rx: watch::Receiver>, + paused: watch::Sender, + paused_rx: watch::Receiver, _entry: Arc, } @@ -58,10 +60,50 @@ impl Subscription { self.rx.borrow().clone() } - /// Wait for the next snapshot. `None` means the subscription ended. + /// Stop delivering snapshots to this subscriber. The shared watch task + /// keeps running; [`next`](Self::next) returns the latest snapshot as soon + /// as the subscription is resumed. + pub fn set_paused(&self, paused: bool) { + self.paused.send_replace(paused); + } + + /// Wait for the next snapshot relevant to this subscriber. `None` means the + /// subscription ended. While paused, snapshots are coalesced and only the + /// latest one is returned on resume. pub async fn next(&mut self) -> Option> { - self.rx.changed().await.ok()?; - Some(self.rx.borrow_and_update().clone()) + let mut paused = *self.paused_rx.borrow_and_update(); + loop { + if paused { + tokio::select! { + result = self.paused_rx.changed() => { + result.ok()?; + if !*self.paused_rx.borrow_and_update() { + return Some(self.rx.borrow_and_update().clone()); + } + } + result = self.rx.changed() => { + result.ok()?; + self.rx.borrow_and_update(); + } + } + } else { + tokio::select! { + result = self.paused_rx.changed() => { + result.ok()?; + paused = *self.paused_rx.borrow_and_update(); + } + result = self.rx.changed() => { + result.ok()?; + if *self.paused_rx.borrow() { + self.rx.borrow_and_update(); + paused = true; + } else { + return Some(self.rx.borrow_and_update().clone()); + } + } + } + } + } } /// Ask the shared watch task to perform a fresh list. @@ -107,8 +149,17 @@ pub(crate) fn subscribe_kind( .retain(|_, entry| entry.strong_count() > 0); if let Some(entry) = repository.entries.get(&key).and_then(Weak::upgrade) { + tracing::trace!( + ?kind, + ?scope, + active = repository.entries.len(), + "repository: subscription reused" + ); + let (paused, paused_rx) = watch::channel(false); return Subscription { rx: entry.snapshots.subscribe(), + paused, + paused_rx, _entry: entry, }; } @@ -139,12 +190,25 @@ pub(crate) fn subscribe_kind( refresh, prefs, }); + let scope_log = scope.clone(); let watch_task = resources::watcher::run( client, kind, scope, key.epoch, snapshots, cancel_rx, refresh_rx, prefs_rx, ); tokio.spawn(watch_task); repository.entries.insert(key, Arc::downgrade(&entry)); - Subscription { rx, _entry: entry } + tracing::debug!( + ?kind, + scope = ?scope_log, + active = repository.entries.len(), + "repository: subscription created" + ); + let (paused, paused_rx) = watch::channel(false); + Subscription { + rx, + paused, + paused_rx, + _entry: entry, + } })) } @@ -186,3 +250,50 @@ fn update_watch_prefs(cx: &mut App, kind: ResourceKind, change: impl FnOnce(&mut } }); } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::*; + use crate::models::SnapshotPhase; + + #[tokio::test] + async fn a_paused_subscription_coalesces_until_resumed() { + let (snapshots, rx) = watch::channel(Arc::new(ResourceSnapshot::loading())); + let (cancel, _) = watch::channel(false); + let (refresh, _) = watch::channel(0u64); + let (prefs, _) = watch::channel(WatchPrefs::default()); + let entry = Arc::new(Entry { + snapshots: snapshots.clone(), + _cancel: cancel, + refresh, + prefs, + }); + let (paused, paused_rx) = watch::channel(false); + let mut subscription = Subscription { + rx, + paused, + paused_rx, + _entry: entry, + }; + + subscription.set_paused(true); + snapshots.send_replace(Arc::new(ResourceSnapshot::failed("offline"))); + let resume = subscription.clone(); + let mut next = Box::pin(subscription.next()); + assert!( + tokio::time::timeout(Duration::from_millis(20), &mut next) + .await + .is_err(), + "a paused subscription must not deliver snapshots" + ); + + resume.set_paused(false); + let snapshot = tokio::time::timeout(Duration::from_millis(250), next) + .await + .expect("resume must wake the pump") + .expect("resume snapshot"); + assert!(matches!(snapshot.phase(), SnapshotPhase::Error(_))); + } +} diff --git a/src/repository/resources/detail.rs b/src/repository/resources/detail.rs index 8953377..e8b76a9 100644 --- a/src/repository/resources/detail.rs +++ b/src/repository/resources/detail.rs @@ -1,23 +1,69 @@ -//! Generic one-shot detail fetch for every Kubernetes resource kind. +//! Live detail follow for every Kubernetes resource kind. -use kube::api::{Api, DynamicObject}; +use std::collections::HashMap; +use std::sync::{Arc, Mutex, OnceLock, Weak}; +use std::time::Duration; + +use futures::TryStreamExt as _; +use kube::ResourceExt; +use kube::api::{Api, DynamicObject, WatchEvent, WatchParams}; use kube::core::{ApiResource, GroupVersionKind}; use serde_json::Value; -use tokio::sync::{mpsc, watch}; +use tokio::sync::{broadcast, watch}; use crate::cluster::ClusterState; use crate::models::ResourceKind; use crate::models::resources::detail::{ResourceDetail, ResourceDetailUpdate}; +use crate::repository::retry::{self, Retry, RetryAction}; use gpui_kit::App; +const DETAIL_WATCH_TIMEOUT_SECS: u32 = 280; +const DETAIL_RECONNECT_DELAY: Duration = Duration::from_millis(250); +const DETAIL_POLL_INTERVAL: Duration = Duration::from_secs(5); +const DETAIL_BROADCAST_CAPACITY: usize = 8; + +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +struct DetailKey { + epoch: u64, + kind: ResourceKind, + namespace: String, + name: String, +} + +struct DetailEntry { + updates: broadcast::Sender, + latest: Arc>>, + _cancel: watch::Sender, +} + +fn detail_registry() -> &'static Mutex>> { + static REGISTRY: OnceLock>>> = OnceLock::new(); + REGISTRY.get_or_init(|| Mutex::new(HashMap::new())) +} + pub struct ResourceDetailSubscription { - updates: mpsc::UnboundedReceiver, + updates: broadcast::Receiver, + initial: Option, cancel: watch::Sender, + cancel_rx: watch::Receiver, + _entry: Arc, } impl ResourceDetailSubscription { pub async fn next(&mut self) -> Option { - self.updates.recv().await + if let Some(initial) = self.initial.take() { + return Some(initial); + } + loop { + tokio::select! { + _ = self.cancel_rx.changed() => return None, + result = self.updates.recv() => match result { + Ok(update) => return Some(update), + Err(broadcast::error::RecvError::Lagged(_)) => continue, + Err(broadcast::error::RecvError::Closed) => return None, + } + } + } } pub fn cancel_handle(&self) -> watch::Sender { @@ -40,17 +86,64 @@ pub fn subscribe_detail( let state = cx.global::(); let client = state.client().ok_or_else(|| "not connected".to_string())?; let tokio = state.tokio(); - let (updates_tx, updates) = mpsc::unbounded_channel(); + let key = DetailKey { + epoch: state.context_epoch(), + kind, + namespace: namespace.to_string(), + name: name.to_string(), + }; + + let mut registry = detail_registry() + .lock() + .map_err(|_| "detail registry is poisoned".to_string())?; + registry.retain(|_, entry| entry.strong_count() > 0); + + if let Some(entry) = registry.get(&key).and_then(Weak::upgrade) { + return Ok(subscription_for(entry)); + } + + let (updates, _) = broadcast::channel(DETAIL_BROADCAST_CAPACITY); let (cancel, cancel_rx) = watch::channel(false); + let latest = Arc::new(Mutex::new(None)); + let entry = Arc::new(DetailEntry { + updates: updates.clone(), + latest: Arc::clone(&latest), + _cancel: cancel, + }); + let receiver = updates.subscribe(); tokio.spawn(watch_detail( client, kind, namespace.to_string(), name.to_string(), - updates_tx, + updates, + latest, cancel_rx, )); - Ok(ResourceDetailSubscription { updates, cancel }) + registry.insert(key, Arc::downgrade(&entry)); + drop(registry); + + let (sub_cancel, sub_cancel_rx) = watch::channel(false); + Ok(ResourceDetailSubscription { + updates: receiver, + initial: None, + cancel: sub_cancel, + cancel_rx: sub_cancel_rx, + _entry: entry, + }) +} + +fn subscription_for(entry: Arc) -> ResourceDetailSubscription { + let updates = entry.updates.subscribe(); + let initial = entry.latest.lock().ok().and_then(|latest| latest.clone()); + let (cancel, cancel_rx) = watch::channel(false); + ResourceDetailSubscription { + updates, + initial, + cancel, + cancel_rx, + _entry: entry, + } } async fn watch_detail( @@ -58,39 +151,301 @@ async fn watch_detail( kind: ResourceKind, namespace: String, name: String, - updates: mpsc::UnboundedSender, + updates: broadcast::Sender, + latest: Arc>>, mut cancel: watch::Receiver, ) { - let resource = api_resource(kind); - let api: Api = if kind.is_namespaced() { - Api::namespaced_with(client, &namespace, &resource) - } else { - Api::all_with(client, &resource) - }; - let fetched = tokio::select! { - _ = cancel.changed() => return, - result = api.get(&name) => result, - }; - let result = match fetched { - Ok(object) => { - let mut value = serde_json::to_value(&object).unwrap_or(Value::Null); - sanitize_secret(kind, &mut value); - Ok(Box::new(ResourceDetail { - kind, - namespace: namespace.clone(), - name: name.clone(), - raw_json: serde_json::to_string_pretty(&value).unwrap_or_default(), - value, - })) + let mut retry = Retry::new(); + let mut resource_version = String::new(); + let mut need_get = true; + + loop { + let api = api_for(&client, kind, &namespace); + if need_get { + let fetched = tokio::select! { + _ = cancel.changed() => return, + result = api.get(&name) => result, + }; + match fetched { + Ok(object) => { + retry.list_succeeded(); + if let Some(version) = object.resource_version() { + resource_version = version; + } + if !send_detail(&updates, &latest, kind, &namespace, &name, &object) { + return; + } + need_get = false; + } + Err(error) => { + if !publish( + &updates, + &latest, + ResourceDetailUpdate { + kind, + namespace: namespace.clone(), + name: name.clone(), + result: Err(format!( + "Could not load {} details: {error}", + kind.label() + )), + }, + ) { + return; + } + match retry.failed(&error) { + RetryAction::Retry(delay) => { + if !sleep_or_cancel(&mut cancel, delay).await { + return; + } + continue; + } + RetryAction::GiveUp => return, + } + } + } } - Err(error) => Err(format!("Could not load {} details: {error}", kind.label())), - }; - let _ = updates.send(ResourceDetailUpdate { + + let params = WatchParams::default() + .timeout(DETAIL_WATCH_TIMEOUT_SECS) + .fields(&format!("metadata.name={name}")); + let watched = tokio::select! { + _ = cancel.changed() => return, + result = retry::bounded(api.watch(¶ms, &resource_version)) => result, + }; + let mut stream = match watched { + Ok(stream) => { + retry.watch_succeeded(); + Box::pin(stream) + } + Err(failure) => match failure.action(&mut retry) { + RetryAction::Retry(delay) => { + if !sleep_or_cancel(&mut cancel, delay).await { + return; + } + continue; + } + RetryAction::GiveUp => { + poll_detail( + &client, + kind, + &namespace, + &name, + &updates, + &latest, + &mut cancel, + ) + .await; + return; + } + }, + }; + + let mut relist = false; + loop { + tokio::select! { + _ = cancel.changed() => return, + event = stream.try_next() => match event { + Ok(Some(WatchEvent::Added(object) | WatchEvent::Modified(object))) => { + retry.watch_succeeded(); + if let Some(version) = object.resource_version() { + resource_version = version; + } + if !send_detail(&updates, &latest, kind, &namespace, &name, &object) { + return; + } + } + Ok(Some(WatchEvent::Deleted(_))) => { + publish( + &updates, + &latest, + ResourceDetailUpdate { + kind, + namespace: namespace.clone(), + name: name.clone(), + result: Err(format!("{} {} was deleted", kind.label(), name)), + }, + ); + return; + } + Ok(Some(WatchEvent::Bookmark(bookmark))) => { + retry.watch_succeeded(); + resource_version = bookmark.metadata.resource_version; + } + Ok(Some(WatchEvent::Error(status))) => { + relist = status.code == 410 + || status.reason == "Expired" + || status.reason == "Gone"; + match retry.failed_status(status.code) { + RetryAction::Retry(delay) => { + if !sleep_or_cancel(&mut cancel, delay).await { + return; + } + break; + } + RetryAction::GiveUp => { + poll_detail( + &client, + kind, + &namespace, + &name, + &updates, + &latest, + &mut cancel, + ) + .await; + return; + } + } + } + Ok(None) => break, + Err(error) => match retry.failed(&error) { + RetryAction::Retry(delay) => { + if !sleep_or_cancel(&mut cancel, delay).await { + return; + } + break; + } + RetryAction::GiveUp => { + poll_detail( + &client, + kind, + &namespace, + &name, + &updates, + &latest, + &mut cancel, + ) + .await; + return; + } + }, + } + } + } + + if relist { + need_get = true; + } + if !sleep_or_cancel(&mut cancel, DETAIL_RECONNECT_DELAY).await { + return; + } + } +} + +/// Watch was denied (or the object is not watchable): keep the detail fresh +/// with a low-frequency GET instead of hammering the watch endpoint. +async fn poll_detail( + client: &kube::Client, + kind: ResourceKind, + namespace: &str, + name: &str, + updates: &broadcast::Sender, + latest: &Arc>>, + cancel: &mut watch::Receiver, +) { + let mut retry = Retry::new(); + let api = api_for(client, kind, namespace); + loop { + let fetched = tokio::select! { + _ = cancel.changed() => return, + result = api.get(name) => result, + }; + match fetched { + Ok(object) => { + retry.list_succeeded(); + if !send_detail(updates, latest, kind, namespace, name, &object) { + return; + } + } + Err(error) => { + if !publish( + updates, + latest, + ResourceDetailUpdate { + kind, + namespace: namespace.to_string(), + name: name.to_string(), + result: Err(format!("Could not load {} details: {error}", kind.label())), + }, + ) { + return; + } + match retry.failed(&error) { + RetryAction::Retry(delay) => { + if !sleep_or_cancel(cancel, delay).await { + return; + } + continue; + } + RetryAction::GiveUp => return, + } + } + } + if !sleep_or_cancel(cancel, DETAIL_POLL_INTERVAL).await { + return; + } + } +} + +fn fmt_detail( + kind: ResourceKind, + namespace: &str, + name: &str, + object: &DynamicObject, +) -> ResourceDetailUpdate { + let mut value = serde_json::to_value(object).unwrap_or(Value::Null); + sanitize_secret(kind, &mut value); + ResourceDetailUpdate { kind, - namespace, - name, - result, - }); + namespace: namespace.to_string(), + name: name.to_string(), + result: Ok(Box::new(ResourceDetail { + kind, + namespace: namespace.to_string(), + name: name.to_string(), + raw_json: serde_json::to_string_pretty(&value).unwrap_or_default(), + value, + })), + } +} + +fn send_detail( + updates: &broadcast::Sender, + latest: &Arc>>, + kind: ResourceKind, + namespace: &str, + name: &str, + object: &DynamicObject, +) -> bool { + publish(updates, latest, fmt_detail(kind, namespace, name, object)) +} + +fn publish( + updates: &broadcast::Sender, + latest: &Arc>>, + update: ResourceDetailUpdate, +) -> bool { + if let Ok(mut latest) = latest.lock() { + *latest = Some(update.clone()); + } + updates.send(update).is_ok() +} + +async fn sleep_or_cancel(cancel: &mut watch::Receiver, duration: Duration) -> bool { + tokio::select! { + _ = cancel.changed() => false, + _ = tokio::time::sleep(duration) => true, + } +} + +fn api_for(client: &kube::Client, kind: ResourceKind, namespace: &str) -> Api { + let resource = api_resource(kind); + if kind.is_namespaced() { + Api::namespaced_with(client.clone(), namespace, &resource) + } else { + Api::all_with(client.clone(), &resource) + } } fn api_resource(kind: ResourceKind) -> ApiResource { diff --git a/src/repository/resources/nodes/mod.rs b/src/repository/resources/nodes/mod.rs index 62b7aa2..df63642 100644 --- a/src/repository/resources/nodes/mod.rs +++ b/src/repository/resources/nodes/mod.rs @@ -3,8 +3,9 @@ use std::collections::BTreeMap; use std::time::Duration; +use futures::TryStreamExt as _; use k8s_openapi::api::core::v1::Node; -use kube::api::{Api, DynamicObject, ListParams}; +use kube::api::{Api, DynamicObject, ListParams, WatchEvent, WatchParams}; use kube::core::{ApiResource, GroupVersionKind}; use kube::{Client, ResourceExt}; use tokio::sync::{mpsc, watch}; @@ -15,11 +16,14 @@ use crate::models::resources::node::detail::{ NodeUsage, }; use crate::models::{NamespaceScope, ResourceKind}; +use crate::repository::retry::{self, Retry, RetryAction}; use gpui_kit::App; use crate::repository::{Subscription, subscribe_kind}; const METRICS_POLL_INTERVAL: Duration = Duration::from_secs(10); +const DETAIL_WATCH_TIMEOUT_SECS: u32 = 280; +const DETAIL_RECONNECT_DELAY: Duration = Duration::from_millis(250); /// Subscribe to the live Nodes stream. /// Returns a [`Subscription`] that delivers immutable snapshots as the cluster changes. @@ -160,59 +164,241 @@ async fn watch_node_detail( mut cancel: watch::Receiver, ) { let api: Api = Api::all(client.clone()); - let fetched = tokio::select! { - _ = cancel.changed() => return, - result = api.get(&node) => result, - }; - let fetched = match fetched { - Ok(node) => node, - Err(error) => { - let _ = updates.send(NodeDetailUpdate::Detail { - node, - result: Err(format!("Could not load node details: {error}")), - }); + let mut retry = Retry::new(); + let mut resource_version = String::new(); + let mut need_get = true; + let mut cpu_capacity = None; + let mut memory_capacity = None; + let mut ticker = tokio::time::interval(METRICS_POLL_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + + loop { + if need_get { + let fetched = tokio::select! { + _ = cancel.changed() => return, + result = api.get(&node) => result, + }; + let object = match fetched { + Ok(object) => object, + Err(error) => { + let _ = updates.send(NodeDetailUpdate::Detail { + node, + result: Err(format!("Could not load node details: {error}")), + }); + return; + } + }; + retry.list_succeeded(); + if let Some(version) = object.resource_version() { + resource_version = version; + } + let detail = detail_from_node(&object); + (cpu_capacity, memory_capacity) = capacity_from_detail(&detail); + if !send_node_detail(&updates, &node, detail) { + return; + } + need_get = false; + } + + let params = WatchParams::default() + .timeout(DETAIL_WATCH_TIMEOUT_SECS) + .fields(&format!("metadata.name={node}")); + let watched = tokio::select! { + _ = cancel.changed() => return, + result = retry::bounded(api.watch(¶ms, &resource_version)) => result, + }; + let mut stream = match watched { + Ok(stream) => { + retry.watch_succeeded(); + Box::pin(stream) + } + Err(failure) => match failure.action(&mut retry) { + RetryAction::Retry(delay) => { + if !sleep_or_cancel(&mut cancel, delay).await { + return; + } + continue; + } + RetryAction::GiveUp => { + poll_node_detail(&client, &node, &updates, &mut cancel).await; + return; + } + }, + }; + + let mut relist = false; + loop { + tokio::select! { + _ = cancel.changed() => return, + _ = ticker.tick() => { + let usage = fetch_usage(&client, &node, cpu_capacity, memory_capacity).await; + if updates + .send(NodeDetailUpdate::Usage { + node: node.clone(), + result: usage, + }) + .is_err() + { + return; + } + } + event = stream.try_next() => match event { + Ok(Some(WatchEvent::Added(object) | WatchEvent::Modified(object))) => { + retry.watch_succeeded(); + if let Some(version) = object.resource_version() { + resource_version = version; + } + let detail = detail_from_node(&object); + (cpu_capacity, memory_capacity) = capacity_from_detail(&detail); + if !send_node_detail(&updates, &node, detail) { + return; + } + } + Ok(Some(WatchEvent::Deleted(_))) => { + let _ = updates.send(NodeDetailUpdate::Detail { + node, + result: Err("This node was deleted".to_string()), + }); + return; + } + Ok(Some(WatchEvent::Bookmark(bookmark))) => { + retry.watch_succeeded(); + resource_version = bookmark.metadata.resource_version; + } + Ok(Some(WatchEvent::Error(status))) => { + relist = status.code == 410 + || status.reason == "Expired" + || status.reason == "Gone"; + match retry.failed_status(status.code) { + RetryAction::Retry(delay) => { + if !sleep_or_cancel(&mut cancel, delay).await { + return; + } + break; + } + RetryAction::GiveUp => { + poll_node_detail(&client, &node, &updates, &mut cancel).await; + return; + } + } + } + Ok(None) => break, + Err(error) => match retry.failed(&error) { + RetryAction::Retry(delay) => { + if !sleep_or_cancel(&mut cancel, delay).await { + return; + } + break; + } + RetryAction::GiveUp => { + poll_node_detail(&client, &node, &updates, &mut cancel).await; + return; + } + }, + } + } + } + + if relist { + need_get = true; + } + if !sleep_or_cancel(&mut cancel, DETAIL_RECONNECT_DELAY).await { return; } - }; - let detail = detail_from_node(&fetched); - let cpu_capacity = detail + } +} + +/// Watch is not permitted: keep both the object and its usage fresh with GETs. +async fn poll_node_detail( + client: &Client, + node: &str, + updates: &mpsc::UnboundedSender, + cancel: &mut watch::Receiver, +) { + let api: Api = Api::all(client.clone()); + let mut retry = Retry::new(); + loop { + let fetched = tokio::select! { + _ = cancel.changed() => return, + result = api.get(node) => result, + }; + match fetched { + Ok(object) => { + retry.list_succeeded(); + let detail = detail_from_node(&object); + let (cpu_capacity, memory_capacity) = capacity_from_detail(&detail); + if !send_node_detail(updates, node, detail) { + return; + } + let usage = fetch_usage(client, node, cpu_capacity, memory_capacity).await; + if updates + .send(NodeDetailUpdate::Usage { + node: node.to_string(), + result: usage, + }) + .is_err() + { + return; + } + } + Err(error) => { + if updates + .send(NodeDetailUpdate::Detail { + node: node.to_string(), + result: Err(format!("Could not load node details: {error}")), + }) + .is_err() + { + return; + } + match retry.failed(&error) { + RetryAction::Retry(delay) => { + if !sleep_or_cancel(cancel, delay).await { + return; + } + continue; + } + RetryAction::GiveUp => return, + } + } + } + if !sleep_or_cancel(cancel, METRICS_POLL_INTERVAL).await { + return; + } + } +} + +fn capacity_from_detail(detail: &NodeDetail) -> (Option, Option) { + let cpu = detail .allocatable .get("cpu") .and_then(|quantity| parse_quantity(quantity)) .map(|cores| (cores * 1000.0).round() as u64); - let memory_capacity = detail + let memory = detail .allocatable .get("memory") .and_then(|quantity| parse_quantity(quantity)) .map(|bytes| bytes.max(0.0).round() as u64); - if updates + (cpu, memory) +} + +fn send_node_detail( + updates: &mpsc::UnboundedSender, + node: &str, + detail: NodeDetail, +) -> bool { + updates .send(NodeDetailUpdate::Detail { - node: node.clone(), + node: node.to_string(), result: Ok(Box::new(detail)), }) - .is_err() - { - return; - } + .is_ok() +} - loop { - let fetched = tokio::select! { - _ = cancel.changed() => return, - result = fetch_usage(&client, &node, cpu_capacity, memory_capacity) => result, - }; - if updates - .send(NodeDetailUpdate::Usage { - node: node.clone(), - result: fetched, - }) - .is_err() - { - return; - } - tokio::select! { - _ = cancel.changed() => return, - _ = tokio::time::sleep(METRICS_POLL_INTERVAL) => {} - } +async fn sleep_or_cancel(cancel: &mut watch::Receiver, duration: Duration) -> bool { + tokio::select! { + _ = cancel.changed() => false, + _ = tokio::time::sleep(duration) => true, } } diff --git a/src/repository/resources/pods/mod.rs b/src/repository/resources/pods/mod.rs index c64162b..32b87a6 100644 --- a/src/repository/resources/pods/mod.rs +++ b/src/repository/resources/pods/mod.rs @@ -7,7 +7,9 @@ use futures::{AsyncBufReadExt, SinkExt, TryStreamExt}; use k8s_openapi::api::core::v1::{ Container, ContainerState, ContainerStatus, EnvVar, Pod, PodStatus, Toleration, Volume, }; -use kube::api::{Api, AttachParams, DynamicObject, LogParams, TerminalSize}; +use kube::api::{ + Api, AttachParams, DynamicObject, LogParams, TerminalSize, WatchEvent, WatchParams, +}; use kube::core::{ApiResource, GroupVersionKind}; use kube::{Client, ResourceExt}; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; @@ -21,10 +23,12 @@ use crate::models::resources::pod::detail::{ use crate::models::{NamespaceScope, ResourceKind}; use gpui_kit::App; -use crate::repository::retry::{Retry, RetryAction}; +use crate::repository::retry::{self, Retry, RetryAction}; use crate::repository::{Subscription, subscribe_kind}; const METRICS_POLL_INTERVAL: Duration = Duration::from_secs(10); +const DETAIL_WATCH_TIMEOUT_SECS: u32 = 280; +const DETAIL_RECONNECT_DELAY: Duration = Duration::from_millis(250); /// Subscribe to the live Pods stream. /// Returns a [`Subscription`] that delivers immutable snapshots as the cluster changes. @@ -462,56 +466,237 @@ async fn watch_pod_detail( mut cancel: watch::Receiver, ) { let api: Api = Api::namespaced(client.clone(), &namespace); - let fetched = tokio::select! { - _ = cancel.changed() => return, - result = api.get(&name) => result, - }; - let fetched = match fetched { - Ok(pod) => pod, - Err(error) => { - let _ = updates.send(PodDetailUpdate::Detail { - namespace, - name, - result: Err(format!("Could not load pod details: {error}")), - }); + let mut retry = Retry::new(); + let mut resource_version = String::new(); + let mut need_get = true; + let mut cpu_capacity = None; + let mut memory_capacity = None; + let mut ticker = tokio::time::interval(METRICS_POLL_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + + loop { + if need_get { + let fetched = tokio::select! { + _ = cancel.changed() => return, + result = api.get(&name) => result, + }; + let pod = match fetched { + Ok(pod) => pod, + Err(error) => { + let _ = updates.send(PodDetailUpdate::Detail { + namespace, + name, + result: Err(format!("Could not load pod details: {error}")), + }); + return; + } + }; + retry.list_succeeded(); + if let Some(version) = pod.resource_version() { + resource_version = version; + } + (cpu_capacity, memory_capacity) = capacity_from_pod(&pod); + if !send_pod_detail(&updates, &namespace, &name, &pod) { + return; + } + need_get = false; + } + + let params = WatchParams::default() + .timeout(DETAIL_WATCH_TIMEOUT_SECS) + .fields(&format!("metadata.name={name}")); + let watched = tokio::select! { + _ = cancel.changed() => return, + result = retry::bounded(api.watch(¶ms, &resource_version)) => result, + }; + let mut stream = match watched { + Ok(stream) => { + retry.watch_succeeded(); + Box::pin(stream) + } + Err(failure) => match failure.action(&mut retry) { + RetryAction::Retry(delay) => { + if !sleep_or_cancel(&mut cancel, delay).await { + return; + } + continue; + } + RetryAction::GiveUp => { + poll_pod_detail(&client, &namespace, &name, &updates, &mut cancel).await; + return; + } + }, + }; + + let mut relist = false; + loop { + tokio::select! { + _ = cancel.changed() => return, + _ = ticker.tick() => { + let usage = fetch_usage(&client, &namespace, &name, cpu_capacity, memory_capacity).await; + if updates + .send(PodDetailUpdate::Usage { + namespace: namespace.clone(), + name: name.clone(), + result: usage, + }) + .is_err() + { + return; + } + } + event = stream.try_next() => match event { + Ok(Some(WatchEvent::Added(pod) | WatchEvent::Modified(pod))) => { + retry.watch_succeeded(); + if let Some(version) = pod.resource_version() { + resource_version = version; + } + (cpu_capacity, memory_capacity) = capacity_from_pod(&pod); + if !send_pod_detail(&updates, &namespace, &name, &pod) { + return; + } + } + Ok(Some(WatchEvent::Deleted(_))) => { + let _ = updates.send(PodDetailUpdate::Detail { + namespace, + name, + result: Err("This pod was deleted".to_string()), + }); + return; + } + Ok(Some(WatchEvent::Bookmark(bookmark))) => { + retry.watch_succeeded(); + resource_version = bookmark.metadata.resource_version; + } + Ok(Some(WatchEvent::Error(status))) => { + relist = status.code == 410 + || status.reason == "Expired" + || status.reason == "Gone"; + match retry.failed_status(status.code) { + RetryAction::Retry(delay) => { + if !sleep_or_cancel(&mut cancel, delay).await { + return; + } + break; + } + RetryAction::GiveUp => { + poll_pod_detail(&client, &namespace, &name, &updates, &mut cancel) + .await; + return; + } + } + } + Ok(None) => break, + Err(error) => match retry.failed(&error) { + RetryAction::Retry(delay) => { + if !sleep_or_cancel(&mut cancel, delay).await { + return; + } + break; + } + RetryAction::GiveUp => { + poll_pod_detail(&client, &namespace, &name, &updates, &mut cancel).await; + return; + } + }, + } + } + } + + if relist { + need_get = true; + } + if !sleep_or_cancel(&mut cancel, DETAIL_RECONNECT_DELAY).await { return; } - }; - let detail = detail_from_pod(&fetched); - let (cpu_capacity, memory_capacity) = capacity_from_pod(&fetched); - if updates - .send(PodDetailUpdate::Detail { - namespace: namespace.clone(), - name: name.clone(), - result: Ok(Box::new(detail)), - }) - .is_err() - { - return; } +} +/// Watch is not permitted: keep both the object and its usage fresh with GETs. +async fn poll_pod_detail( + client: &Client, + namespace: &str, + name: &str, + updates: &mpsc::UnboundedSender, + cancel: &mut watch::Receiver, +) { + let api: Api = Api::namespaced(client.clone(), namespace); + let mut retry = Retry::new(); loop { let fetched = tokio::select! { _ = cancel.changed() => return, - result = fetch_usage(&client, &namespace, &name, cpu_capacity, memory_capacity) => result, + result = api.get(name) => result, }; - if updates - .send(PodDetailUpdate::Usage { - namespace: namespace.clone(), - name: name.clone(), - result: fetched, - }) - .is_err() - { - return; + match fetched { + Ok(pod) => { + retry.list_succeeded(); + let (cpu_capacity, memory_capacity) = capacity_from_pod(&pod); + if !send_pod_detail(updates, namespace, name, &pod) { + return; + } + let usage = + fetch_usage(client, namespace, name, cpu_capacity, memory_capacity).await; + if updates + .send(PodDetailUpdate::Usage { + namespace: namespace.to_string(), + name: name.to_string(), + result: usage, + }) + .is_err() + { + return; + } + } + Err(error) => { + if updates + .send(PodDetailUpdate::Detail { + namespace: namespace.to_string(), + name: name.to_string(), + result: Err(format!("Could not load pod details: {error}")), + }) + .is_err() + { + return; + } + match retry.failed(&error) { + RetryAction::Retry(delay) => { + if !sleep_or_cancel(cancel, delay).await { + return; + } + continue; + } + RetryAction::GiveUp => return, + } + } } - tokio::select! { - _ = cancel.changed() => return, - _ = tokio::time::sleep(METRICS_POLL_INTERVAL) => {} + if !sleep_or_cancel(cancel, METRICS_POLL_INTERVAL).await { + return; } } } +fn send_pod_detail( + updates: &mpsc::UnboundedSender, + namespace: &str, + name: &str, + pod: &Pod, +) -> bool { + updates + .send(PodDetailUpdate::Detail { + namespace: namespace.to_string(), + name: name.to_string(), + result: Ok(Box::new(detail_from_pod(pod))), + }) + .is_ok() +} + +async fn sleep_or_cancel(cancel: &mut watch::Receiver, duration: Duration) -> bool { + tokio::select! { + _ = cancel.changed() => false, + _ = tokio::time::sleep(duration) => true, + } +} + async fn fetch_usage( client: &Client, namespace: &str, diff --git a/src/repository/resources/watcher.rs b/src/repository/resources/watcher.rs index 27a00a1..8b781a9 100644 --- a/src/repository/resources/watcher.rs +++ b/src/repository/resources/watcher.rs @@ -4,7 +4,7 @@ //! [`ResourceStore`] and only publishes immutable snapshots to the UI, throttled //! to `FLUSH_INTERVAL` or every `FLUSH_AT` events, whichever comes first. -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::fmt::Debug; use std::sync::{Arc, Mutex, OnceLock}; use std::time::Duration; @@ -39,10 +39,70 @@ use crate::repository::watch_prefs::{WatchPreference, WatchPrefs}; const FLUSH_INTERVAL: Duration = Duration::from_millis(50); const FLUSH_AT: usize = 256; +/// How often an automatically denied watch is retried. RBAC can change without +/// an app restart, but re-probing too often would spam the API server. +const WATCH_REPROBE_INTERVAL: Duration = Duration::from_secs(600); + +/// Streaming lists (`sendInitialEvents`) bounds. A server that ignores the +/// option never sends the initial-end bookmark, so an idle stream falls back to +/// the list path quickly. A busy initial stream gets longer, but is capped. +const STREAM_INITIAL_IDLE_TIMEOUT: Duration = Duration::from_secs(5); +const STREAM_INITIAL_TOTAL_TIMEOUT: Duration = Duration::from_secs(60); + +/// Kubernetes version at which `sendInitialEvents` was introduced. 1.27 and +/// 1.28 may still have the WatchList feature gate disabled, so the stream is +/// still attempted and falls back to list if no initial-end bookmark arrives. +const STREAM_INITIAL_MIN_MINOR: u32 = 27; + +static STREAMING_SUPPORT: OnceLock>> = OnceLock::new(); + +fn streaming_support() -> &'static Mutex> { + STREAMING_SUPPORT.get_or_init(|| Mutex::new(HashMap::new())) +} + +async fn supports_streaming_initial(client: &Client, epoch: u64) -> bool { + if let Ok(cache) = streaming_support().lock() + && let Some(supported) = cache.get(&epoch) + { + return *supported; + } + + let supported = match retry::bounded(client.apiserver_version()).await { + Ok(info) => version_supports_streaming(&info.major, &info.minor), + Err(failure) => { + tracing::debug!(%failure, "repository: could not determine API server version"); + false + } + }; + if let Ok(mut cache) = streaming_support().lock() { + cache.insert(epoch, supported); + } + supported +} + +fn version_supports_streaming(major: &str, minor: &str) -> bool { + let major = major.parse::().unwrap_or(0); + let minor = minor + .trim_end_matches(|character: char| !character.is_ascii_digit()) + .parse::() + .unwrap_or(0); + major > 1 || (major == 1 && minor >= STREAM_INITIAL_MIN_MINOR) +} + +fn bookmark_ends_initial_events(annotations: &std::collections::BTreeMap) -> bool { + annotations + .get("k8s.io/initial-events-end") + .is_some_and(|value| value == "true") +} + /// List chunk size. Big kinds (thousands of pods/events) render their first /// chunk immediately instead of waiting for one huge response. const LIST_CHUNK_SIZE: u32 = 500; +/// Events are high-churn and grow without bound while the watch is open. +/// Keep the newest rows, matching Freelens' 1000-event initial list. +const EVENTS_LIMIT: usize = 1_000; + /// Kinds whose watch the API server denied (RBAC) for the current context /// epoch. New subscriptions for these kinds skip the doomed watch attempt and /// start polling right away, so changing the namespace filter does not reset @@ -96,6 +156,8 @@ pub async fn run( prefs, store: ResourceStore::new(), mode: WatchMode::Watch, + ready: false, + stream_initial_disabled: false, }; match kind { ResourceKind::Pods => task.run_namespaced::(scope, pod_item).await, @@ -196,6 +258,11 @@ struct WatchTask { prefs: watch::Receiver, store: ResourceStore, mode: WatchMode, + /// Whether a `Ready` snapshot has been published for this task. Used to + /// publish an empty first list without republishing unchanged poll results. + ready: bool, + /// Once streaming lists fall back, keep using the list path for this task. + stream_initial_disabled: bool, } /// Why the poll loop exited. @@ -209,13 +276,30 @@ enum PollOutcome { /// Outcome of a chunked list. enum ListOutcome { /// Full list published; watch from this resource version. - Listed(String), + Listed { + resource_version: String, + changed: bool, + }, /// Subscription cancelled mid-list. Cancelled, /// A chunk request failed or timed out; the caller decides whether to retry. Failed(RequestFailure), } +/// Outcome of attempting a Kubernetes streaming list. +enum InitialStream { + /// Initial events and the end-of-initial-events bookmark were received. + Listed { + resource_version: String, + changed: bool, + }, + /// Subscription cancelled. + Cancelled, + /// The server ignored/rejected streaming lists, the stream ended, or the + /// initial bookmark did not arrive in time. + Unsupported, +} + impl WatchTask { async fn run_namespaced(&mut self, scope: NamespaceScope, map: F) where @@ -282,28 +366,68 @@ impl WatchTask { } } if need_list { - match self.list_progressively(&api, map).await { - ListOutcome::Listed(version) => { - retry.list_succeeded(); - resource_version = version; - need_list = false; + let mut streamed = false; + if !self.stream_initial_disabled + && !self.ready + && self.store.len() == 0 + && !watch_denied(self.epoch, self.kind) + && supports_streaming_initial(&self.client, self.epoch).await + { + match self.stream_initial(&api, map).await { + InitialStream::Listed { + resource_version: version, + changed, + } => { + tracing::debug!( + changed, + "repository: initial state came from a streaming list" + ); + retry.list_succeeded(); + resource_version = version; + need_list = false; + streamed = true; + } + InitialStream::Cancelled => return, + InitialStream::Unsupported => { + tracing::debug!( + "repository: streaming list unsupported or timed out, falling back to list" + ); + self.stream_initial_disabled = true; + } } - ListOutcome::Cancelled => return, - ListOutcome::Failed(failure) => match failure.action(&mut retry) { - RetryAction::Retry(delay) => { - tracing::warn!(%failure, ?api, delay_ms = delay.as_millis() as u64, "repository: list failed, retrying"); - self.publish(SnapshotPhase::Error(format!("list failed: {failure}"))); - if !self.sleep(delay).await { + } + + if !streamed { + match self.list_progressively(&api, map).await { + ListOutcome::Listed { + resource_version: version, + .. + } => { + retry.list_succeeded(); + resource_version = version; + need_list = false; + } + ListOutcome::Cancelled => return, + ListOutcome::Failed(failure) => match failure.action(&mut retry) { + RetryAction::Retry(delay) => { + tracing::warn!(%failure, ?api, delay_ms = delay.as_millis() as u64, "repository: list failed, retrying"); + self.publish(SnapshotPhase::Error(format!( + "list failed: {failure}" + ))); + if !self.sleep(delay).await { + return; + } + continue; + } + RetryAction::GiveUp => { + tracing::error!(%failure, ?api, "repository: list rejected, stopping watch"); + self.publish(SnapshotPhase::Error(format!( + "list rejected: {failure}" + ))); return; } - continue; - } - RetryAction::GiveUp => { - tracing::error!(%failure, ?api, "repository: list rejected, stopping watch"); - self.publish(SnapshotPhase::Error(format!("list rejected: {failure}"))); - return; - } - }, + }, + } } } @@ -456,6 +580,7 @@ impl WatchTask { } }, } + self.trim_high_churn(); if self.store.pending() >= FLUSH_AT { self.publish(SnapshotPhase::Ready); } @@ -474,11 +599,18 @@ impl WatchTask { } } + fn trim_high_churn(&mut self) -> bool { + if self.kind == ResourceKind::Events { + return self.store.trim_oldest(EVENTS_LIMIT); + } + false + } + fn publish(&mut self, phase: SnapshotPhase) { + self.ready = matches!(&phase, SnapshotPhase::Ready); let snapshot = Arc::new(self.store.snapshot(phase).with_mode(self.mode)); tracing::trace!( objects = self.store.len(), - revision = snapshot.revision(), "repository: publishing snapshot" ); self.tx.send_replace(snapshot); @@ -492,8 +624,131 @@ impl WatchTask { } } - /// List every object in chunks, publishing each chunk so large kinds - /// render progressively instead of waiting for one huge response. + /// Try Kubernetes 1.27+ streaming lists (`sendInitialEvents`) before the + /// chunked list path. Returns `Unsupported` for old API servers, disabled + /// feature gates, RBAC errors, or a stream that never emits the + /// `k8s.io/initial-events-end` bookmark. The caller then falls back to the + /// existing list+watch code. + async fn stream_initial(&mut self, api: &Api, map: F) -> InitialStream + where + K: Resource + Clone + Debug + DeserializeOwned + Send + 'static, + F: Fn(&K) -> Item + Copy, + { + let params = WatchParams::streaming_lists().timeout(280); + let watched = tokio::select! { + _ = self.cancel.changed() => return InitialStream::Cancelled, + result = retry::bounded(api.watch(¶ms, "")) => result, + }; + let mut stream = match watched { + Ok(stream) => Box::pin(stream), + Err(failure) => { + tracing::debug!(%failure, ?api, "repository: streaming list setup failed"); + return InitialStream::Unsupported; + } + }; + + let mut resource_version = String::new(); + let mut changed = false; + let total_deadline = tokio::time::Instant::now() + STREAM_INITIAL_TOTAL_TIMEOUT; + let mut idle_deadline = tokio::time::Instant::now() + STREAM_INITIAL_IDLE_TIMEOUT; + let mut ticker = tokio::time::interval(FLUSH_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + + loop { + let deadline = total_deadline.min(idle_deadline); + tokio::select! { + _ = self.cancel.changed() => return InitialStream::Cancelled, + _ = ticker.tick() => { + if self.store.pending() > 0 { + self.publish(SnapshotPhase::Ready); + } + } + _ = tokio::time::sleep_until(deadline) => { + tracing::debug!( + ?api, + "repository: streaming list did not finish, falling back to list" + ); + return InitialStream::Unsupported; + } + event = stream.try_next() => match event { + Ok(Some(WatchEvent::Added(object) | WatchEvent::Modified(object))) => { + idle_deadline = + tokio::time::Instant::now() + STREAM_INITIAL_IDLE_TIMEOUT; + let item = map(&object); + if !item.resource_version.is_empty() { + resource_version.clone_from(&item.resource_version); + } + changed |= self.store.upsert(item); + changed |= self.trim_high_churn(); + if self.store.pending() >= FLUSH_AT { + self.publish(SnapshotPhase::Ready); + } + } + Ok(Some(WatchEvent::Deleted(object))) => { + idle_deadline = + tokio::time::Instant::now() + STREAM_INITIAL_IDLE_TIMEOUT; + if let Some(version) = object.resource_version() { + resource_version = version; + } + changed |= self.store.remove(&ObjectKey::new( + object.namespace().unwrap_or_default(), + object.name_any(), + )); + changed |= self.trim_high_churn(); + } + Ok(Some(WatchEvent::Bookmark(bookmark))) => { + let initial_end = + bookmark_ends_initial_events(&bookmark.metadata.annotations); + if !bookmark.metadata.resource_version.is_empty() { + resource_version = bookmark.metadata.resource_version; + } + if !initial_end { + tracing::debug!( + ?api, + "repository: streaming list bookmark was not the initial-events-end" + ); + return InitialStream::Unsupported; + } + if self.store.pending() > 0 || !self.ready { + self.publish(SnapshotPhase::Ready); + } + return InitialStream::Listed { + resource_version, + changed, + }; + } + Ok(Some(WatchEvent::Error(status))) => { + tracing::debug!( + code = status.code, + message = %status.message, + ?api, + "repository: streaming list error, falling back to list" + ); + return InitialStream::Unsupported; + } + Ok(None) => { + tracing::debug!( + ?api, + "repository: streaming list ended early, falling back to list" + ); + return InitialStream::Unsupported; + } + Err(error) => { + tracing::debug!( + %error, + ?api, + "repository: streaming list failed, falling back to list" + ); + return InitialStream::Unsupported; + } + } + } + } + } + + /// Merge every object in the list into the store, publishing each chunk so + /// large kinds render progressively. Objects missing from the completed + /// list are removed at the end; unchanged objects are left untouched. /// Returns the resource version to watch from. async fn list_progressively(&mut self, api: &Api, map: F) -> ListOutcome where @@ -502,29 +757,34 @@ impl WatchTask { { let mut params = ListParams::default().limit(LIST_CHUNK_SIZE); let mut resource_version = String::new(); - let mut first_chunk = true; + let mut changed = false; + self.store.begin_sync(); loop { let listed = tokio::select! { - _ = self.cancel.changed() => return ListOutcome::Cancelled, + _ = self.cancel.changed() => { + self.store.abort_sync(); + return ListOutcome::Cancelled; + } result = retry::bounded(api.list(¶ms)) => result, }; let chunk = match listed { Ok(chunk) => chunk, - Err(failure) => return ListOutcome::Failed(failure), + Err(failure) => { + self.store.abort_sync(); + return ListOutcome::Failed(failure); + } }; // Every chunk carries the version of the same list snapshot. if let Some(version) = chunk.metadata.resource_version { resource_version = version; } - if first_chunk { - self.store.reset(chunk.items.iter().map(map)); - first_chunk = false; - } else { - for object in &chunk.items { - self.store.upsert(map(object)); - } + for object in &chunk.items { + changed |= self.store.upsert(map(object)); + } + changed |= self.trim_high_churn(); + if self.store.pending() > 0 { + self.publish(SnapshotPhase::Ready); } - self.publish(SnapshotPhase::Ready); match chunk.metadata.continue_ { Some(token) if !token.is_empty() => { params = params.continue_token(&token); @@ -532,7 +792,14 @@ impl WatchTask { _ => break, } } - ListOutcome::Listed(resource_version) + changed |= self.store.finish_sync(); + if self.store.pending() > 0 || !self.ready { + self.publish(SnapshotPhase::Ready); + } + ListOutcome::Listed { + resource_version, + changed, + } } /// Poll instead of watching — because watch was denied (RBAC) or the user @@ -551,6 +818,10 @@ impl WatchTask { self.publish(SnapshotPhase::Ready); } let mut retry = Retry::new(); + let mut unchanged_polls: u32 = 0; + let mut reprobe = tokio::time::interval(WATCH_REPROBE_INTERVAL); + reprobe.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + reprobe.tick().await; loop { if self.cancel.has_changed().is_err() { @@ -558,8 +829,13 @@ impl WatchTask { } match self.list_progressively(api, map).await { - ListOutcome::Listed(_) => { + ListOutcome::Listed { changed, .. } => { retry.list_succeeded(); + if changed { + unchanged_polls = 0; + } else { + unchanged_polls = unchanged_polls.saturating_add(1); + } } ListOutcome::Cancelled => return PollOutcome::Stop, ListOutcome::Failed(failure) => match failure.action(&mut retry) { @@ -580,16 +856,20 @@ impl WatchTask { } // Wait for the next tick. Refresh or a preference change re-list - // immediately; `Off` waits without a timer. + // immediately; `Off` waits without a timer. Repeated unchanged + // polls back off up to one minute. let rate = self.prefs.borrow().rate; + let delay = poll_delay(rate, unchanged_polls); tokio::select! { _ = self.cancel.changed() => return PollOutcome::Stop, _ = self.refresh.changed() => { self.refresh.borrow_and_update(); + unchanged_polls = 0; } _ = self.prefs.changed() => { let prefs = *self.prefs.borrow_and_update(); self.mode = WatchMode::Poll(prefs.rate); + unchanged_polls = 0; if prefs.mode == WatchPreference::Watch { self.mode = WatchMode::Watch; self.publish(SnapshotPhase::Ready); @@ -599,12 +879,23 @@ impl WatchTask { // happens on the next loop turn. self.publish(SnapshotPhase::Ready); } + _ = reprobe.tick() => { + // Only an automatic denial is re-probed; an explicit + // polling preference is left alone. + if self.prefs.borrow().mode == WatchPreference::Watch + && watch_denied(self.epoch, self.kind) + { + tracing::info!(?api, "repository: re-probing previously denied watch"); + clear_watch_denied(self.epoch, self.kind); + self.mode = WatchMode::Watch; + self.publish(SnapshotPhase::Ready); + return PollOutcome::ResumeWatch; + } + } _ = async { - match rate { - PollRate::Every(secs) => { - tokio::time::sleep(Duration::from_secs(secs)).await - } - PollRate::Off => std::future::pending().await, + match delay { + Some(delay) => tokio::time::sleep(delay).await, + None => std::future::pending().await, } } => {} } @@ -612,6 +903,18 @@ impl WatchTask { } } +fn poll_delay(rate: PollRate, unchanged_polls: u32) -> Option { + match rate { + PollRate::Off => None, + PollRate::Every(secs) => { + let factor = 1u64 << unchanged_polls.min(3); + Some(Duration::from_secs( + secs.saturating_mul(factor).clamp(1, 60), + )) + } + } +} + fn base_item(object: &K) -> Item where K: ResourceExt, @@ -1189,4 +1492,49 @@ mod tests { clear_watch_denied(epoch, ResourceKind::Pods); assert!(!watch_denied(epoch, ResourceKind::Pods)); } + + #[test] + fn poll_backoff_doubles_until_the_cap() { + assert_eq!( + poll_delay(PollRate::Every(5), 0), + Some(Duration::from_secs(5)) + ); + assert_eq!( + poll_delay(PollRate::Every(5), 1), + Some(Duration::from_secs(10)) + ); + assert_eq!( + poll_delay(PollRate::Every(5), 2), + Some(Duration::from_secs(20)) + ); + assert_eq!( + poll_delay(PollRate::Every(5), 3), + Some(Duration::from_secs(40)) + ); + assert_eq!( + poll_delay(PollRate::Every(10), 4), + Some(Duration::from_secs(60)) + ); + assert_eq!(poll_delay(PollRate::Off, 3), None); + } + + #[test] + fn streaming_lists_require_kubernetes_1_27() { + assert!(!version_supports_streaming("1", "26")); + assert!(version_supports_streaming("1", "27")); + assert!(version_supports_streaming("1", "37")); + assert!(version_supports_streaming("1", "37+")); + assert!(version_supports_streaming("2", "0")); + assert!(!version_supports_streaming("0", "99")); + } + + #[test] + fn only_the_initial_events_bookmark_ends_the_stream() { + let mut annotations = std::collections::BTreeMap::new(); + assert!(!bookmark_ends_initial_events(&annotations)); + annotations.insert("k8s.io/initial-events-end".into(), "false".into()); + assert!(!bookmark_ends_initial_events(&annotations)); + annotations.insert("k8s.io/initial-events-end".into(), "true".into()); + assert!(bookmark_ends_initial_events(&annotations)); + } } diff --git a/src/repository/watch_prefs.rs b/src/repository/watch_prefs.rs index 89e2cc8..d5d4029 100644 --- a/src/repository/watch_prefs.rs +++ b/src/repository/watch_prefs.rs @@ -42,12 +42,13 @@ pub struct WatchPrefsStore { kinds: HashMap, } -/// The factory preference for a kind. Namespaces default to polling: many -/// clusters deny `watch` on them and they change rarely. Picking "Live watch" -/// in the UI saves and sticks across launches. +/// The factory preference for a kind. Namespaces, Events, and Leases default +/// to polling: many clusters deny `watch` on them and they are high-churn or +/// rarely change. Picking "Live watch" in the UI saves and sticks across +/// launches. fn default_prefs_for(kind: ResourceKind) -> WatchPrefs { match kind { - ResourceKind::Namespaces => WatchPrefs { + ResourceKind::Namespaces | ResourceKind::Events | ResourceKind::Leases => WatchPrefs { mode: WatchPreference::Poll, rate: PollRate::DEFAULT, }, @@ -153,15 +154,21 @@ mod tests { } #[test] - fn namespaces_default_to_polling_until_watch_is_chosen() { + fn high_churn_kinds_default_to_polling_until_watch_is_chosen() { let mut store = WatchPrefsStore::default(); - assert_eq!( - store.prefs_for(ResourceKind::Namespaces), - WatchPrefs { - mode: WatchPreference::Poll, - rate: PollRate::Every(5), - } - ); + for kind in [ + ResourceKind::Namespaces, + ResourceKind::Events, + ResourceKind::Leases, + ] { + assert_eq!( + store.prefs_for(kind), + WatchPrefs { + mode: WatchPreference::Poll, + rate: PollRate::Every(5), + } + ); + } // Choosing live watch is saved and sticks. store.update(ResourceKind::Namespaces, |prefs| { diff --git a/src/ui/resources/list.rs b/src/ui/resources/list.rs index 8118614..f6fcb61 100644 --- a/src/ui/resources/list.rs +++ b/src/ui/resources/list.rs @@ -123,6 +123,8 @@ impl ResourceListView { pub fn set_active(&mut self, active: bool, cx: &mut Context) { self.vm.set_active(active, cx); + self.detail + .update(cx, |detail, cx| detail.set_active(active, cx)); } pub fn close_detail(&mut self, cx: &mut Context) { diff --git a/src/ui/shell/mod.rs b/src/ui/shell/mod.rs index fe1336d..1b92177 100644 --- a/src/ui/shell/mod.rs +++ b/src/ui/shell/mod.rs @@ -620,7 +620,8 @@ impl Workspace { } /// Only the active tab of each visible group is live. Hidden tabs keep - /// their retained views but stop receiving watch updates until shown again. + /// their retained views and shared watch, but pause snapshot delivery and + /// detail/metrics work until shown again. fn sync_visibility(&mut self, cx: &mut Context) { let visible = self .shell diff --git a/src/view_models/resources/cluster_roles/mod.rs b/src/view_models/resources/cluster_roles/mod.rs index a2b9130..cad01bc 100644 --- a/src/view_models/resources/cluster_roles/mod.rs +++ b/src/view_models/resources/cluster_roles/mod.rs @@ -48,6 +48,10 @@ impl ClusterRolesViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -114,8 +118,26 @@ impl ClusterRolesViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -153,6 +175,7 @@ impl ClusterRolesViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -164,7 +187,6 @@ impl ClusterRolesViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -175,12 +197,9 @@ impl ClusterRolesViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/config_maps/mod.rs b/src/view_models/resources/config_maps/mod.rs index 49be129..84b1834 100644 --- a/src/view_models/resources/config_maps/mod.rs +++ b/src/view_models/resources/config_maps/mod.rs @@ -44,6 +44,10 @@ impl ConfigMapsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -110,8 +114,26 @@ impl ConfigMapsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -149,6 +171,7 @@ impl ConfigMapsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -160,7 +183,6 @@ impl ConfigMapsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -171,12 +193,9 @@ impl ConfigMapsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/cron_jobs/mod.rs b/src/view_models/resources/cron_jobs/mod.rs index 7a948be..1e0e0c7 100644 --- a/src/view_models/resources/cron_jobs/mod.rs +++ b/src/view_models/resources/cron_jobs/mod.rs @@ -43,6 +43,10 @@ impl CronJobsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -109,8 +113,26 @@ impl CronJobsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -148,6 +170,7 @@ impl CronJobsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -159,7 +182,6 @@ impl CronJobsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -170,12 +192,9 @@ impl CronJobsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/custom.rs b/src/view_models/resources/custom.rs index 74ebb31..ec86f73 100644 --- a/src/view_models/resources/custom.rs +++ b/src/view_models/resources/custom.rs @@ -54,6 +54,10 @@ impl CustomResourceViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -116,8 +120,26 @@ impl CustomResourceViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -153,6 +175,7 @@ impl CustomResourceViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -164,7 +187,6 @@ impl CustomResourceViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -175,12 +197,9 @@ impl CustomResourceViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/daemon_sets/mod.rs b/src/view_models/resources/daemon_sets/mod.rs index 6d68e71..8b27628 100644 --- a/src/view_models/resources/daemon_sets/mod.rs +++ b/src/view_models/resources/daemon_sets/mod.rs @@ -44,6 +44,10 @@ impl DaemonSetsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -110,8 +114,26 @@ impl DaemonSetsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -149,6 +171,7 @@ impl DaemonSetsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -160,7 +183,6 @@ impl DaemonSetsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -171,12 +193,9 @@ impl DaemonSetsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/deployments/mod.rs b/src/view_models/resources/deployments/mod.rs index e3fe0c4..7318046 100644 --- a/src/view_models/resources/deployments/mod.rs +++ b/src/view_models/resources/deployments/mod.rs @@ -52,6 +52,10 @@ impl DeploymentsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -118,8 +122,26 @@ impl DeploymentsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -157,6 +179,7 @@ impl DeploymentsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -168,7 +191,6 @@ impl DeploymentsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -179,12 +201,9 @@ impl DeploymentsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/detail.rs b/src/view_models/resources/detail.rs index c8d2803..f9d56b5 100644 --- a/src/view_models/resources/detail.rs +++ b/src/view_models/resources/detail.rs @@ -11,6 +11,7 @@ use gpui_kit::Context; pub struct ResourceDetailViewModel { state: ResourceDetailState, epoch: u64, + active: bool, cancel: Option>, #[allow(dead_code)] // held for its Drop side-effect: cancels the GPUI observer cluster_sub: gpui_kit::Subscription, @@ -32,6 +33,7 @@ impl ResourceDetailViewModel { Self { state: ResourceDetailState::Closed, epoch, + active: true, cancel: None, cluster_sub, } @@ -41,6 +43,25 @@ impl ResourceDetailViewModel { &self.state } + /// Cancel the live detail stream while the owning tab is hidden, and + /// resume it without resetting the visible detail when the tab returns. + pub fn set_active(&mut self, active: bool, cx: &mut Context) { + if self.active == active { + return; + } + self.active = active; + if !active { + self.cancel_detail(); + return; + } + if let Some((kind, namespace, name)) = self.state.identity() { + let (kind, namespace, name) = (kind, namespace.to_string(), name.to_string()); + if self.cancel.is_none() { + self.subscribe(kind, namespace, name, cx); + } + } + } + pub fn open( &mut self, kind: ResourceKind, @@ -60,6 +81,18 @@ impl ResourceDetailViewModel { name: name.clone(), }; cx.notify(); + if self.active { + self.subscribe(kind, namespace, name, cx); + } + } + + fn subscribe( + &mut self, + kind: ResourceKind, + namespace: String, + name: String, + cx: &mut Context, + ) { let mut subscription = match repository::resources::detail::subscribe_detail(cx, kind, &namespace, &name) { Ok(subscription) => subscription, diff --git a/src/view_models/resources/endpoint_slices/mod.rs b/src/view_models/resources/endpoint_slices/mod.rs index ec0a9e8..566a6be 100644 --- a/src/view_models/resources/endpoint_slices/mod.rs +++ b/src/view_models/resources/endpoint_slices/mod.rs @@ -48,6 +48,10 @@ impl EndpointSlicesViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -114,8 +118,26 @@ impl EndpointSlicesViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -153,6 +175,7 @@ impl EndpointSlicesViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -164,7 +187,6 @@ impl EndpointSlicesViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -175,12 +197,9 @@ impl EndpointSlicesViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/endpoints/mod.rs b/src/view_models/resources/endpoints/mod.rs index a1e4602..6e8cffb 100644 --- a/src/view_models/resources/endpoints/mod.rs +++ b/src/view_models/resources/endpoints/mod.rs @@ -43,6 +43,10 @@ impl EndpointsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -109,8 +113,26 @@ impl EndpointsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -148,6 +170,7 @@ impl EndpointsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -159,7 +182,6 @@ impl EndpointsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -170,12 +192,9 @@ impl EndpointsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/events/mod.rs b/src/view_models/resources/events/mod.rs index e0561f3..e8a5512 100644 --- a/src/view_models/resources/events/mod.rs +++ b/src/view_models/resources/events/mod.rs @@ -43,6 +43,10 @@ impl EventsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -109,8 +113,26 @@ impl EventsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -147,6 +169,7 @@ impl EventsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -158,7 +181,6 @@ impl EventsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -169,12 +191,9 @@ impl EventsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/horizontal_pod_autoscalers/mod.rs b/src/view_models/resources/horizontal_pod_autoscalers/mod.rs index 90bcc0d..d1959c6 100644 --- a/src/view_models/resources/horizontal_pod_autoscalers/mod.rs +++ b/src/view_models/resources/horizontal_pod_autoscalers/mod.rs @@ -48,6 +48,10 @@ impl HorizontalPodAutoscalersViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -114,8 +118,26 @@ impl HorizontalPodAutoscalersViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -155,6 +177,7 @@ impl HorizontalPodAutoscalersViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -166,7 +189,6 @@ impl HorizontalPodAutoscalersViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -177,12 +199,9 @@ impl HorizontalPodAutoscalersViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/ingresses/mod.rs b/src/view_models/resources/ingresses/mod.rs index edf0627..0263e39 100644 --- a/src/view_models/resources/ingresses/mod.rs +++ b/src/view_models/resources/ingresses/mod.rs @@ -43,6 +43,10 @@ impl IngressesViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -109,8 +113,26 @@ impl IngressesViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -148,6 +170,7 @@ impl IngressesViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -159,7 +182,6 @@ impl IngressesViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -170,12 +192,9 @@ impl IngressesViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/jobs/mod.rs b/src/view_models/resources/jobs/mod.rs index 32a77f5..350b6ac 100644 --- a/src/view_models/resources/jobs/mod.rs +++ b/src/view_models/resources/jobs/mod.rs @@ -43,6 +43,10 @@ impl JobsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -109,8 +113,26 @@ impl JobsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -147,6 +169,7 @@ impl JobsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -158,7 +181,6 @@ impl JobsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -169,12 +191,9 @@ impl JobsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/leases/mod.rs b/src/view_models/resources/leases/mod.rs index 642c0b8..c4169a1 100644 --- a/src/view_models/resources/leases/mod.rs +++ b/src/view_models/resources/leases/mod.rs @@ -43,6 +43,10 @@ impl LeasesViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -109,8 +113,26 @@ impl LeasesViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -147,6 +169,7 @@ impl LeasesViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -158,7 +181,6 @@ impl LeasesViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -169,12 +191,9 @@ impl LeasesViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/namespaces/mod.rs b/src/view_models/resources/namespaces/mod.rs index f1d864f..bf4aa32 100644 --- a/src/view_models/resources/namespaces/mod.rs +++ b/src/view_models/resources/namespaces/mod.rs @@ -44,6 +44,10 @@ impl NamespacesViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -110,8 +114,26 @@ impl NamespacesViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -148,6 +170,7 @@ impl NamespacesViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -159,7 +182,6 @@ impl NamespacesViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -170,12 +192,9 @@ impl NamespacesViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/network_policies/mod.rs b/src/view_models/resources/network_policies/mod.rs index e558817..7150378 100644 --- a/src/view_models/resources/network_policies/mod.rs +++ b/src/view_models/resources/network_policies/mod.rs @@ -48,6 +48,10 @@ impl NetworkPoliciesViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -114,8 +118,26 @@ impl NetworkPoliciesViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -153,6 +175,7 @@ impl NetworkPoliciesViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -164,7 +187,6 @@ impl NetworkPoliciesViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -175,12 +197,9 @@ impl NetworkPoliciesViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/node/mod.rs b/src/view_models/resources/node/mod.rs index a3f6150..ed61fa1 100644 --- a/src/view_models/resources/node/mod.rs +++ b/src/view_models/resources/node/mod.rs @@ -54,6 +54,11 @@ impl NodesViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + this.cancel_detail(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -134,8 +139,30 @@ impl NodesViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } + if let Some(node) = self.current_detail_node(cx) { + self.attach_detail(node, cx); + } cx.notify(); + } else { + self.cancel_detail(); } } @@ -205,6 +232,13 @@ impl NodesViewModel { model.begin_detail(node.clone()); cx.notify(); }); + self.attach_detail(node, cx); + } + + /// Restart the detail stream without resetting the selected detail state, + /// used when a hidden tab becomes visible again. + fn attach_detail(&mut self, node: String, cx: &mut Context) { + self.cancel_detail(); let mut detail_sub = match repository::resources::nodes::subscribe_detail(cx, &node) { Ok(subscription) => subscription, Err(error) => { @@ -257,6 +291,7 @@ impl NodesViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -268,7 +303,6 @@ impl NodesViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -279,12 +313,9 @@ impl NodesViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/overview.rs b/src/view_models/resources/overview.rs index 31fdd6f..7fae225 100644 --- a/src/view_models/resources/overview.rs +++ b/src/view_models/resources/overview.rs @@ -43,6 +43,11 @@ impl ClusterOverviewViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + this.cancel_metrics(); + return; + } this.subscribe(cx); this.subscribe_metrics(cx); }); @@ -86,14 +91,39 @@ impl ClusterOverviewViewModel { self.metrics_error.as_deref() } + pub fn is_active(&self) -> bool { + self.active.load(Ordering::Relaxed) + } + pub fn set_active(&mut self, active: bool, cx: &mut Context) { let was_active = self.active.swap(active, Ordering::Relaxed); if was_active == active { return; } + if let Some(nodes) = &self.nodes { + nodes.set_paused(!active); + } + if let Some(pods) = &self.pods { + pods.set_paused(!active); + } if active { + if self.nodes.is_none() || self.pods.is_none() { + self.subscribe(cx); + self.subscribe_metrics(cx); + cx.notify(); + return; + } + if let Some(nodes) = &self.nodes { + self.node_snapshot = nodes.snapshot(); + } + if let Some(pods) = &self.pods { + self.pod_snapshot = pods.snapshot(); + } + self.subscribe_metrics(cx); self.recompute(); cx.notify(); + } else { + self.cancel_metrics(); } } @@ -130,9 +160,11 @@ impl ClusterOverviewViewModel { } }; if let Some(nodes) = &self.nodes { + nodes.set_paused(!self.is_active()); self.node_snapshot = nodes.snapshot(); } if let Some(pods) = &self.pods { + pods.set_paused(!self.is_active()); self.pod_snapshot = pods.snapshot(); } self.recompute(); @@ -141,7 +173,6 @@ impl ClusterOverviewViewModel { self.pump_cancel = Some(cancel); let mut nodes = self.nodes.clone(); let mut pods = self.pods.clone(); - let active = Arc::clone(&self.active); cx.spawn(async move |this, cx| { loop { tokio::select! { @@ -151,13 +182,10 @@ impl ClusterOverviewViewModel { nodes = None; continue; }; - let should_notify = active.load(Ordering::Relaxed); let applied = this.update(cx, |this, cx| { this.node_snapshot = snapshot; this.recompute(); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; @@ -168,13 +196,10 @@ impl ClusterOverviewViewModel { pods = None; continue; }; - let should_notify = active.load(Ordering::Relaxed); let applied = this.update(cx, |this, cx| { this.pod_snapshot = snapshot; this.recompute(); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; @@ -200,7 +225,20 @@ impl ClusterOverviewViewModel { self.pods = None; } + fn cancel_metrics(&mut self) { + if let Some(cancel) = self.metrics_cancel.take() { + let _ = cancel.send(true); + } + self.metrics.clear(); + self.metrics_error = None; + self.metrics_generation = self.metrics_generation.wrapping_add(1); + } + fn subscribe_metrics(&mut self, cx: &mut Context) { + if !self.is_active() { + self.cancel_metrics(); + return; + } if let Some(cancel) = self.metrics_cancel.take() { let _ = cancel.send(true); } diff --git a/src/view_models/resources/persistent_volume_claims/mod.rs b/src/view_models/resources/persistent_volume_claims/mod.rs index 7ce2e32..eef76f3 100644 --- a/src/view_models/resources/persistent_volume_claims/mod.rs +++ b/src/view_models/resources/persistent_volume_claims/mod.rs @@ -48,6 +48,10 @@ impl PersistentVolumeClaimsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -114,8 +118,26 @@ impl PersistentVolumeClaimsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -155,6 +177,7 @@ impl PersistentVolumeClaimsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -166,7 +189,6 @@ impl PersistentVolumeClaimsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -177,12 +199,9 @@ impl PersistentVolumeClaimsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/persistent_volumes/mod.rs b/src/view_models/resources/persistent_volumes/mod.rs index 008785d..7e9adeb 100644 --- a/src/view_models/resources/persistent_volumes/mod.rs +++ b/src/view_models/resources/persistent_volumes/mod.rs @@ -48,6 +48,10 @@ impl PersistentVolumesViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -114,8 +118,26 @@ impl PersistentVolumesViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -152,6 +174,7 @@ impl PersistentVolumesViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -163,7 +186,6 @@ impl PersistentVolumesViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -174,12 +196,9 @@ impl PersistentVolumesViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/pod/mod.rs b/src/view_models/resources/pod/mod.rs index 1499138..1128dd6 100644 --- a/src/view_models/resources/pod/mod.rs +++ b/src/view_models/resources/pod/mod.rs @@ -57,6 +57,11 @@ impl PodsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + this.cancel_detail(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -137,8 +142,30 @@ impl PodsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } + if let Some((namespace, name)) = self.current_detail_identity(cx) { + self.attach_detail(namespace, name, cx); + } cx.notify(); + } else { + self.cancel_detail(); } } @@ -214,6 +241,13 @@ impl PodsViewModel { model.begin_detail(namespace.clone(), name.clone()); cx.notify(); }); + self.attach_detail(namespace, name, cx); + } + + /// Restart the detail stream without resetting the selected detail state, + /// used when a hidden tab becomes visible again. + fn attach_detail(&mut self, namespace: String, name: String, cx: &mut Context) { + self.cancel_detail(); let mut detail_sub = match repository::resources::pods::subscribe_detail(cx, &namespace, &name) { Ok(subscription) => subscription, @@ -266,6 +300,7 @@ impl PodsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -277,7 +312,6 @@ impl PodsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -288,12 +322,9 @@ impl PodsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/pod_disruption_budgets/mod.rs b/src/view_models/resources/pod_disruption_budgets/mod.rs index cb432a2..ef52404 100644 --- a/src/view_models/resources/pod_disruption_budgets/mod.rs +++ b/src/view_models/resources/pod_disruption_budgets/mod.rs @@ -48,6 +48,10 @@ impl PodDisruptionBudgetsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -114,8 +118,26 @@ impl PodDisruptionBudgetsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -155,6 +177,7 @@ impl PodDisruptionBudgetsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -166,7 +189,6 @@ impl PodDisruptionBudgetsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -177,12 +199,9 @@ impl PodDisruptionBudgetsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/replica_sets/mod.rs b/src/view_models/resources/replica_sets/mod.rs index d1cbfa8..87aeaa7 100644 --- a/src/view_models/resources/replica_sets/mod.rs +++ b/src/view_models/resources/replica_sets/mod.rs @@ -48,6 +48,10 @@ impl ReplicaSetsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -114,8 +118,26 @@ impl ReplicaSetsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -153,6 +175,7 @@ impl ReplicaSetsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -164,7 +187,6 @@ impl ReplicaSetsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -175,12 +197,9 @@ impl ReplicaSetsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/replication_controllers/mod.rs b/src/view_models/resources/replication_controllers/mod.rs index 0aeb064..0d46a2c 100644 --- a/src/view_models/resources/replication_controllers/mod.rs +++ b/src/view_models/resources/replication_controllers/mod.rs @@ -48,6 +48,10 @@ impl ReplicationControllersViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -114,8 +118,26 @@ impl ReplicationControllersViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -154,6 +176,7 @@ impl ReplicationControllersViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -165,7 +188,6 @@ impl ReplicationControllersViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -176,12 +198,9 @@ impl ReplicationControllersViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/role_bindings/mod.rs b/src/view_models/resources/role_bindings/mod.rs index d1b9ea2..f86597c 100644 --- a/src/view_models/resources/role_bindings/mod.rs +++ b/src/view_models/resources/role_bindings/mod.rs @@ -48,6 +48,10 @@ impl RoleBindingsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -114,8 +118,26 @@ impl RoleBindingsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -153,6 +175,7 @@ impl RoleBindingsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -164,7 +187,6 @@ impl RoleBindingsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -175,12 +197,9 @@ impl RoleBindingsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/roles/mod.rs b/src/view_models/resources/roles/mod.rs index d3b315b..07d22cc 100644 --- a/src/view_models/resources/roles/mod.rs +++ b/src/view_models/resources/roles/mod.rs @@ -43,6 +43,10 @@ impl RolesViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -109,8 +113,26 @@ impl RolesViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -147,6 +169,7 @@ impl RolesViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -158,7 +181,6 @@ impl RolesViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -169,12 +191,9 @@ impl RolesViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/secrets/mod.rs b/src/view_models/resources/secrets/mod.rs index 03f6baa..c5cd142 100644 --- a/src/view_models/resources/secrets/mod.rs +++ b/src/view_models/resources/secrets/mod.rs @@ -43,6 +43,10 @@ impl SecretsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -109,8 +113,26 @@ impl SecretsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -147,6 +169,7 @@ impl SecretsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -158,7 +181,6 @@ impl SecretsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -169,12 +191,9 @@ impl SecretsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/service_accounts/mod.rs b/src/view_models/resources/service_accounts/mod.rs index e8e99ed..a7ce13e 100644 --- a/src/view_models/resources/service_accounts/mod.rs +++ b/src/view_models/resources/service_accounts/mod.rs @@ -48,6 +48,10 @@ impl ServiceAccountsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -114,8 +118,26 @@ impl ServiceAccountsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -153,6 +175,7 @@ impl ServiceAccountsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -164,7 +187,6 @@ impl ServiceAccountsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -175,12 +197,9 @@ impl ServiceAccountsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/services/mod.rs b/src/view_models/resources/services/mod.rs index 6f2e518..11a3c52 100644 --- a/src/view_models/resources/services/mod.rs +++ b/src/view_models/resources/services/mod.rs @@ -43,6 +43,10 @@ impl ServicesViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -109,8 +113,26 @@ impl ServicesViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -148,6 +170,7 @@ impl ServicesViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -159,7 +182,6 @@ impl ServicesViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -170,12 +192,9 @@ impl ServicesViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/stateful_sets/mod.rs b/src/view_models/resources/stateful_sets/mod.rs index 4d6b653..741c40a 100644 --- a/src/view_models/resources/stateful_sets/mod.rs +++ b/src/view_models/resources/stateful_sets/mod.rs @@ -48,6 +48,10 @@ impl StatefulSetsViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -114,8 +118,26 @@ impl StatefulSetsViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -153,6 +175,7 @@ impl StatefulSetsViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -164,7 +187,6 @@ impl StatefulSetsViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -175,12 +197,9 @@ impl StatefulSetsViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break; diff --git a/src/view_models/resources/storage_classes/mod.rs b/src/view_models/resources/storage_classes/mod.rs index 57781e3..00b188e 100644 --- a/src/view_models/resources/storage_classes/mod.rs +++ b/src/view_models/resources/storage_classes/mod.rs @@ -48,6 +48,10 @@ impl StorageClassesViewModel { return; } this.epoch = epoch; + if !this.is_active() { + this.cancel_pump(); + return; + } this.subscribe(cx); cx.notify(); }); @@ -114,8 +118,26 @@ impl StorageClassesViewModel { if was_active == active { return; } + if let Some(subscription) = &self.subscription { + subscription.set_paused(!active); + } if active { - cx.notify(); + if self.subscription.is_none() { + self.subscribe(cx); + return; + } + let snapshot = self + .subscription + .as_ref() + .map(|subscription| subscription.snapshot()); + if let Some(snapshot) = snapshot { + self.model.update(cx, |model, cx| { + model.apply_snapshot(snapshot); + cx.notify(); + }); + } else { + cx.notify(); + } } } @@ -152,6 +174,7 @@ impl StorageClassesViewModel { return; } }; + subscription.set_paused(!self.is_active()); let initial = subscription.snapshot(); let notify_now = self.is_active(); self.model.update(cx, |model, cx| { @@ -163,7 +186,6 @@ impl StorageClassesViewModel { let (cancel, mut cancel_rx) = watch::channel(()); self.pump_cancel = Some(cancel); let model = self.model.downgrade(); - let active = Arc::clone(&self.active); let mut pump = subscription.clone(); cx.spawn(async move |_, cx: &mut gpui_kit::AsyncApp| { loop { @@ -174,12 +196,9 @@ impl StorageClassesViewModel { let Some(snapshot) = snapshot else { break; }; - let should_notify = active.load(Ordering::Relaxed); let applied = model.update(cx, |model, cx| { model.apply_snapshot(snapshot); - if should_notify { - cx.notify(); - } + cx.notify(); }); if applied.is_err() { break;