This commit is contained in:
2026-09-06 20:12:05 +07:00
parent c2c2839ac1
commit 03a239b1ce
2 changed files with 230 additions and 189 deletions
+226 -66
View File
@@ -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<RepoAddr, Option<String>>,
/// 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<Instant>,
tasks: Vec<Task<Result<(), Error>>>,
_subscriptions: Vec<Subscription>,
}
@@ -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>) {
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<RepoAddr, Vec<CheckoutStatus>> = HashMap::new();
for (addr, announced_head) in &requested {
let Some(paths) = associations.get(addr) else {
continue;
};
let list: Vec<CheckoutStatus> = 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<RepoAddr, Vec<CheckoutStatus>> = HashMap::new();
for addr in &push_requested {
let Some(paths) = associations.get(addr) else {
continue;
};
let list: Vec<CheckoutStatus> = 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<Self>) {
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<Self>) {
// 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<Self>) {
let associations = self.by_repo.clone();
let requested: Vec<(RepoAddr, Option<String>)> = self
.status_requested
.iter()
.map(|addr| {
(
addr.clone(),
self.requested_head.get(addr).cloned().flatten(),
)
})
.collect();
let push_requested: Vec<RepoAddr> = 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<CheckoutStatus> {
///
/// `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<CheckoutStatus> {
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<CheckoutStatus> {
} 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<CheckoutStatus> {
})
}
/// 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<RepoAddr, Vec<PathBuf>>,
requested: &[(RepoAddr, Option<String>)],
push_requested: &[RepoAddr],
fetch: bool,
) -> (
HashMap<RepoAddr, Vec<CheckoutStatus>>,
HashMap<RepoAddr, Vec<CheckoutStatus>>,
) {
let mut statuses: HashMap<RepoAddr, Vec<CheckoutStatus>> = HashMap::new();
for (addr, announced_head) in requested {
let Some(paths) = associations.get(addr) else {
continue;
};
let list: Vec<CheckoutStatus> = 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<RepoAddr, Vec<CheckoutStatus>> = HashMap::new();
for addr in push_requested {
let Some(paths) = associations.get(addr) else {
continue;
};
let list: Vec<CheckoutStatus> = 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 {
+4 -123
View File
@@ -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);
}
}