From 03a239b1ce8b50d14ae52fc8a36166621c0abb94 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Sun, 6 Sep 2026 20:12:05 +0700 Subject: [PATCH] update --- crates/signed_state/src/checkouts.rs | 292 +++++++++++++++++++++------ crates/signed_state/src/refresh.rs | 127 +----------- 2 files changed, 230 insertions(+), 189 deletions(-) diff --git a/crates/signed_state/src/checkouts.rs b/crates/signed_state/src/checkouts.rs index d9b6547..fb4abcb 100644 --- a/crates/signed_state/src/checkouts.rs +++ b/crates/signed_state/src/checkouts.rs @@ -1,7 +1,7 @@ use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::process::Command; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use anyhow::Error; use gpui::{App, AppContext, Context, Entity, Global, Subscription, Task}; @@ -18,10 +18,17 @@ use crate::repo_list::RepoListStore; /// Delay between a refresh request and the actual re-computation. const REFRESH_DEBOUNCE: Duration = Duration::from_millis(300); -/// How often the statuses of open repository panels are refreshed. +/// How often the statuses are recomputed against the local refs. +/// +/// A commit lands in a checkout long before the remote reconciliation cadence, +/// so this fast pass surfaces ready-to-push and ready-to-contribute checkouts +/// within a second or two. It reads the tracking refs only, no network. +const LOCAL_POLL: Duration = Duration::from_secs(2); + +/// How often a full pass refreshes the remotes while any repository panel is open. const STATUS_POLL: Duration = Duration::from_secs(15); -/// Background poll interval for the `ready to push` badges of the user's own repositories. +/// Remote refresh interval for the `ready to push` badges of the user's own repositories. const PUSH_POLL: Duration = Duration::from_secs(60); /// Maximum checkouts considered per repository when computing statuses. @@ -92,6 +99,13 @@ pub struct CheckoutsStore { requested_head: HashMap>, /// Refresh coalescing, see [`RefreshGate`]. refresh: RefreshGate, + /// A local status pass timer is pending. + local_pending: bool, + /// When the last full pass (with a remote refresh) completed. + /// + /// The local pass runs a full pass again once this is older than the + /// reconciliation cadence, so remote moves still land. + last_full_sync: Option, tasks: Vec>>, _subscriptions: Vec, } @@ -150,6 +164,8 @@ impl CheckoutsStore { push_statuses: HashMap::new(), requested_head: HashMap::new(), refresh: RefreshGate::default(), + local_pending: false, + last_full_sync: None, tasks: Vec::new(), _subscriptions: subscriptions, }; @@ -295,7 +311,13 @@ impl CheckoutsStore { self.push_task(task); } - /// One resolve and apply cycle, the debounced entry point. + /// One full resolve and apply cycle, the debounced entry point. + /// + /// Re-resolves the associations from the settings, the scan and the + /// announcements, then recomputes the requested statuses against freshly + /// fetched remotes. Full passes run on every input change and on the + /// remote reconciliation cadence ([`Self::local_tick`]); they also restart + /// the fast local pass. fn run_refresh(&mut self, cx: &mut Context) { self.refresh.begin(); @@ -361,35 +383,8 @@ impl CheckoutsStore { .map(|(addr, paths)| (addr, paths.into_iter().filter(|p| p.is_dir()).collect())) .collect(); - let mut statuses: HashMap> = HashMap::new(); - for (addr, announced_head) in &requested { - let Some(paths) = associations.get(addr) else { - continue; - }; - let list: Vec = paths - .iter() - .take(MAX_STATUS_CHECKOUTS) - .filter_map(|path| checkout_status(path, announced_head.as_deref())) - .collect(); - if !list.is_empty() { - statuses.insert(addr.clone(), list); - } - } - - let mut push_statuses: HashMap> = HashMap::new(); - for addr in &push_requested { - let Some(paths) = associations.get(addr) else { - continue; - }; - let list: Vec = paths - .iter() - .take(MAX_STATUS_CHECKOUTS) - .filter_map(|path| checkout_push_status(path)) - .collect(); - if !list.is_empty() { - push_statuses.insert(addr.clone(), list); - } - } + let (statuses, push_statuses) = + compute_statuses(&associations, &requested, &push_requested, true); Ok::<_, Error>((associations, statuses, push_statuses)) }); @@ -399,8 +394,11 @@ impl CheckoutsStore { Ok(results) => results, Err(_) => { // Git reads are best-effort, keep the last results. - return this.update(cx, |this, _cx| { + return this.update(cx, |this, cx| { this.refresh.abort(); + if poll { + this.schedule_local_pass(cx); + } }); } }; @@ -414,13 +412,13 @@ impl CheckoutsStore { this.statuses = statuses; this.push_statuses = push_statuses; - // Poll cycles and identity re-requests recompute the same maps - // over and over. Notify only when something actually changed, - // so observers skip the no-op heartbeats. + // Notify only when something actually changed, so observers + // skip the no-op heartbeats. if associations_changed || statuses_changed || push_statuses_changed { cx.notify(); } + this.last_full_sync = Some(Instant::now()); this.refresh.finish() })?; @@ -428,31 +426,127 @@ impl CheckoutsStore { this.update(cx, |this, cx| this.refresh(cx))?; } - // Keep the statuses current while any repository panel is open. + // Restart the fast local pass so the freshly resolved + // associations drive it. The pass itself decides when the next + // full pass runs. this.update(cx, |this, cx| { - if poll && this.refresh.idle() { - this.refresh.poll(); + if poll { + this.schedule_local_pass(cx); + } + })?; - // Open panels get the fast cadence. - // Each cycle fetches every watched checkout's remote. - let delay = if this.status_requested.is_empty() { - PUSH_POLL - } else { - STATUS_POLL - }; + Ok(()) + })); + } - let task = cx.spawn(async move |this, cx| { - cx.background_executor().timer(delay).await; - this.update(cx, |this, cx| { - // A request that arrived while the poll was pending - // superseded it with its own debounce; skip the stale poll. - if this.refresh.take_poll() { - this.run_refresh(cx); - } - }) - }); + /// Schedule the fast local status pass, unless one is already pending. + /// + /// Every [`LOCAL_POLL`] the pass recomputes the requested statuses against + /// the local refs — no network — so a new commit in a checkout surfaces in + /// a second or two instead of at the next remote reconciliation. + fn schedule_local_pass(&mut self, cx: &mut Context) { + if self.local_pending { + return; + } + self.local_pending = true; - this.push_task(task); + let task = cx.spawn(async move |this, cx| { + cx.background_executor().timer(LOCAL_POLL).await; + this.update(cx, |this, cx| { + this.local_pending = false; + this.local_tick(cx); + }) + }); + + self.push_task(task); + } + + /// The fast local status pass. + /// + /// Recomputes the statuses against the local refs; when the remote + /// reconciliation cadence elapsed, it runs a full pass instead so pushes + /// made elsewhere do not linger as `to push`. + fn local_tick(&mut self, cx: &mut Context) { + // Nothing watched: the chain idles out until a new request restarts it. + if self.status_requested.is_empty() && self.push_requested.is_empty() { + return; + } + + // A full pass or a fresh request covers this tick, skip it. + if self.refresh.running() || self.refresh.debouncing() { + self.schedule_local_pass(cx); + return; + } + + // Open panels get the faster remote cadence. + let cadence = if self.status_requested.is_empty() { + PUSH_POLL + } else { + STATUS_POLL + }; + + let full_due = self + .last_full_sync + .is_none_or(|sync| sync.elapsed() >= cadence); + + if full_due { + self.last_full_sync = Some(Instant::now()); + self.refresh(cx); + } else { + self.run_local_statuses(cx); + } + + self.schedule_local_pass(cx); + } + + /// Recompute the requested statuses against the tracking refs only. + /// + /// The refs were last refreshed by a full pass. Comparing against them is + /// enough to pick up new local commits, and skipping the network keeps + /// this pass cheap enough to run every [`LOCAL_POLL`]. + fn run_local_statuses(&mut self, cx: &mut Context) { + let associations = self.by_repo.clone(); + + let requested: Vec<(RepoAddr, Option)> = self + .status_requested + .iter() + .map(|addr| { + ( + addr.clone(), + self.requested_head.get(addr).cloned().flatten(), + ) + }) + .collect(); + + let push_requested: Vec = self.push_requested.iter().cloned().collect(); + + let work = cx.background_spawn(async move { + let (statuses, push_statuses) = + compute_statuses(&associations, &requested, &push_requested, false); + Ok::<_, Error>((statuses, push_statuses)) + }); + + self.push_task(cx.spawn(async move |this, cx| { + let Ok((statuses, push_statuses)) = work.await else { + // Git reads are best-effort, keep the last results. + return Ok(()); + }; + + this.update(cx, |this, cx| { + // A full pass or a fresh request will apply fresher data + // (the tracking refs move only when a full pass fetches). + if this.refresh.running() || this.refresh.debouncing() { + return; + } + + let statuses_changed = this.statuses != statuses; + let push_statuses_changed = this.push_statuses != push_statuses; + + this.statuses = statuses; + this.push_statuses = push_statuses; + + if statuses_changed || push_statuses_changed { + cx.notify(); } })?; @@ -607,19 +701,27 @@ fn ref_exists(path: &Path, name: &str) -> bool { } /// The `ready to push` status of one checkout of the user's own repository. -fn checkout_push_status(path: &Path) -> Option { +/// +/// `fetch` refreshes the remote heads first, so a full pass sees pushes made +/// elsewhere; the fast local pass skips it and compares against the tracking +/// refs left by the last full pass, which is enough to detect local commits. +fn checkout_push_status(path: &Path, fetch: bool) -> Option { if worktree_dirty(path) { return None; } + let branch = current_branch_of(path)?; let head = signed_git::head_commit_id(path).ok().flatten()?; let origin = signed_git::origin_url(path).ok().flatten()?; - // Refresh the remote heads first. - // Commits made elsewhere or pushed from another machine must not linger as `to push`. - signed_git::fetch_repo_refs(path, &[origin], "+refs/heads/*:refs/remotes/origin/*").ok(); + if fetch { + // Refresh the remote heads first. + // Commits made elsewhere or pushed from another machine must not linger as `to push`. + signed_git::fetch_repo_refs(path, &[origin], "+refs/heads/*:refs/remotes/origin/*").ok(); + } let remote = format!("refs/remotes/origin/{branch}"); + // A branch never fetched or pushed yet compares against the remote HEAD. // The remote HEAD is the fork point in practice. let base = if ref_exists(path, &remote) { @@ -629,7 +731,9 @@ fn checkout_push_status(path: &Path) -> Option { } else { return None; }; + let ahead = commits_ahead(path, &base, &branch); + (ahead > 0).then_some(CheckoutStatus { path: path.to_path_buf(), branch, @@ -639,6 +743,57 @@ fn checkout_push_status(path: &Path) -> Option { }) } +/// Compute the requested statuses against the checkout paths of `associations`. +/// +/// Shared by the full and the local pass. `fetch` refreshes the checkouts' +/// remote heads first, so the full pass sees remote moves; the fast local +/// pass reads the tracking refs only, which is enough to detect local commits. +fn compute_statuses( + associations: &HashMap>, + requested: &[(RepoAddr, Option)], + push_requested: &[RepoAddr], + fetch: bool, +) -> ( + HashMap>, + HashMap>, +) { + let mut statuses: HashMap> = HashMap::new(); + for (addr, announced_head) in requested { + let Some(paths) = associations.get(addr) else { + continue; + }; + + let list: Vec = paths + .iter() + .take(MAX_STATUS_CHECKOUTS) + .filter_map(|path| checkout_status(path, announced_head.as_deref())) + .collect(); + + if !list.is_empty() { + statuses.insert(addr.clone(), list); + } + } + + let mut push_statuses: HashMap> = HashMap::new(); + for addr in push_requested { + let Some(paths) = associations.get(addr) else { + continue; + }; + + let list: Vec = paths + .iter() + .take(MAX_STATUS_CHECKOUTS) + .filter_map(|path| checkout_push_status(path, fetch)) + .collect(); + + if !list.is_empty() { + push_statuses.insert(addr.clone(), list); + } + } + + (statuses, push_statuses) +} + /// Whether the pull request `pr` already proposes the same change as `checkout`. pub fn pr_proposes_checkout( pr: &Event, @@ -913,21 +1068,26 @@ mod tests { }; // A fresh clone has nothing to push. - assert_eq!(checkout_push_status(&checkout), None); + assert_eq!(checkout_push_status(&checkout, true), None); // One local commit, ready to push, counted against the remote. std::fs::write(checkout.join("work.txt"), "x\n").expect("write"); run(&["add", "-A"]); run(&["commit", "-m", "local work"]); - let status = checkout_push_status(&checkout).expect("status"); + let status = checkout_push_status(&checkout, true).expect("status"); assert_eq!(status.branch, "main"); assert_eq!(status.base, "refs/remotes/origin/main"); assert_eq!(status.ahead, 1); assert_eq!(status.head.len(), 40); + // The local-only pass reads the tracking refs, no fetch needed: + // a commit lands locally long before the remote is reconciled. + let local = checkout_push_status(&checkout, false).expect("local status"); + assert_eq!(local.ahead, 1); + // After the push the same commit is on the remote, idle again. run(&["push", "origin", "main"]); - assert_eq!(checkout_push_status(&checkout), None); + assert_eq!(checkout_push_status(&checkout, true), None); // A commit made by someone else on the remote must not count as local work. // It is behind, not ahead. @@ -946,7 +1106,7 @@ mod tests { std::fs::write(remote.join("other.txt"), "y\n").expect("write"); remote_run(&["add", "-A"]); remote_run(&["commit", "-m", "remote work"]); - assert_eq!(checkout_push_status(&checkout), None); + assert_eq!(checkout_push_status(&checkout, true), None); } fn pr_event(author: &str, tags: &[&[&str]]) -> Event { diff --git a/crates/signed_state/src/refresh.rs b/crates/signed_state/src/refresh.rs index 3e9532e..e0aa024 100644 --- a/crates/signed_state/src/refresh.rs +++ b/crates/signed_state/src/refresh.rs @@ -3,13 +3,7 @@ /// [`crate::RepoStore`], [`crate::RepoListStore`] and [`crate::CheckoutsStore`] /// re-query their inputs on a debounce timer with the same policy: /// a request arriving while a run is in flight is folded into a follow-up run, -/// a request arriving while a request debounce is pending is dropped by it. -/// -/// A slow poll cycle (`poll`) is different: it keeps a store's derived data -/// fresh while nothing is happening, but it must never delay a real request. -/// A request arriving while a poll is pending supersedes the poll with its own -/// short debounce, so external events (a push landing, a settings change) -/// propagate promptly instead of waiting out the poll interval. +/// a request arriving while the debounce timer is pending is dropped by it. #[derive(Debug, Default)] pub struct RefreshGate { /// A run is in flight. @@ -18,10 +12,6 @@ pub struct RefreshGate { dirty: bool, /// The debounce timer is pending. debouncing: bool, - /// The pending debounce is a poll cycle, not a request. - /// - /// Polls wait for a quiet moment; requests supersede them. - poll: bool, } /// What a refresh request decided. @@ -44,54 +34,25 @@ impl RefreshGate { self.debouncing } - /// Whether no run is in flight and no timer is pending. - pub fn idle(&self) -> bool { - !self.running && !self.debouncing - } - /// A new refresh request arrived. /// - /// Folded into a follow-up run while one is in flight or a request debounce - /// is pending, superseding a slow poll with the request's own debounce, - /// otherwise starts the debounce timer. + /// Folded into a follow-up run while one is in flight, dropped while the + /// debounce timer is pending, otherwise starts the timer. pub fn request(&mut self) -> RefreshRequest { if self.running { self.dirty = true; RefreshRequest::Fold - } else if self.debouncing && !self.poll { + } else if self.debouncing { RefreshRequest::Fold } else { - // A request supersedes a pending poll: start the short debounce. self.debouncing = true; - self.poll = false; RefreshRequest::Schedule } } - /// A slow poll timer was started without a request. - pub fn poll(&mut self) { - self.debouncing = true; - self.poll = true; - } - - /// A poll timer fired. Whether it is still the scheduled pass and may run. - /// - /// A request that arrived while the poll was pending superseded it with its - /// own debounce, so the stale poll timer is skipped. - pub fn take_poll(&mut self) -> bool { - if self.debouncing && self.poll { - self.debouncing = false; - self.poll = false; - true - } else { - false - } - } - /// The debounce timer fired and the run starts now. pub fn begin(&mut self) { self.debouncing = false; - self.poll = false; self.running = true; } @@ -106,83 +67,3 @@ impl RefreshGate { self.running = false; } } - -#[cfg(test)] -mod tests { - use super::{RefreshGate, RefreshRequest}; - - #[test] - fn request_starts_the_debounce_when_idle() { - let mut gate = RefreshGate::default(); - assert_eq!(gate.request(), RefreshRequest::Schedule); - assert!(gate.debouncing()); - assert!(!gate.idle()); - } - - #[test] - fn request_folds_into_a_request_debounce() { - let mut gate = RefreshGate::default(); - gate.request(); - assert_eq!(gate.request(), RefreshRequest::Fold); - } - - #[test] - fn request_folds_into_a_running_run_and_runs_again() { - let mut gate = RefreshGate::default(); - gate.request(); - gate.begin(); - assert!(gate.running()); - assert_eq!(gate.request(), RefreshRequest::Fold); - assert!(gate.finish()); - assert_eq!(gate.request(), RefreshRequest::Schedule); - } - - #[test] - fn request_supersedes_a_pending_poll() { - let mut gate = RefreshGate::default(); - gate.poll(); - assert!(gate.debouncing()); - - // The request starts its own short debounce instead of waiting out the poll. - assert_eq!(gate.request(), RefreshRequest::Schedule); - assert!(gate.debouncing()); - assert!( - !gate.take_poll(), - "the superseded poll timer must be skipped" - ); - - // The request's own debounce still fires. - gate.begin(); - assert!(gate.running()); - } - - #[test] - fn poll_timer_runs_when_not_superseded() { - let mut gate = RefreshGate::default(); - gate.poll(); - assert!(gate.take_poll()); - assert!(gate.idle()); - gate.begin(); - assert!(gate.running()); - } - - #[test] - fn poll_keeps_scheduling_until_a_request_preempts() { - let mut gate = RefreshGate::default(); - gate.poll(); - assert!(gate.take_poll()); - gate.begin(); - gate.finish(); - gate.poll(); - gate.request(); - assert!(!gate.take_poll(), "preempted by the request"); - } - - #[test] - fn request_after_a_superseded_poll_is_folded_into_the_new_debounce() { - let mut gate = RefreshGate::default(); - gate.poll(); - gate.request(); - assert_eq!(gate.request(), RefreshRequest::Fold); - } -}