1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
//! Workspace-registry and rescan handlers for the [`Broker`](super::Broker). Split out of
//! `daemon.rs` to keep it under the 1000-line `rust-max-lines` cap; this is a second `impl Broker`
//! block and the request dispatch stays in `daemon.rs`.
use super::*;
impl Broker {
/// Scan/rescan a workspace on the sole-writer pool. The scan is CPU-bound, so it runs on a
/// blocking thread while the reactor keeps serving other links. A scan/store error becomes a
/// `CommsResponse::Error` (never a torn link).
pub(super) async fn on_rescan(
&self,
root: std::path::PathBuf,
paths: Option<Vec<std::path::PathBuf>>,
full: bool,
embed: bool,
) -> CommsResponse {
self.mark_active().await;
// A tripped scan token means the daemon is draining (every drain route cancels first, ~keep
// and the token never un-trips). Launching a scan now would only produce a doomed ~keep
// partial pass that defeats coalescing — refuse instead; the client retries against ~keep
// the next daemon. ~keep
if self.scan_cancel.is_cancelled() {
return CommsResponse::Error {
code: "rescan_draining".to_string(),
message: "daemon draining; rescan refused (retry against the next daemon)".to_string(),
};
}
let _rescan_guard = self.blob_gc_lock.read().await;
let pool = Arc::clone(&self.workspaces);
let cancel = self.scan_cancel.clone();
let started = Instant::now();
match tokio::task::spawn_blocking(move || pool.rescan(&root, paths, full, embed, &cancel)).await {
// A cancelled pass committed only part of the tree — surface it as an error so no ~keep
// client mistakes the partial index for a completed rescan. ~keep
Ok(Ok((_, true))) => CommsResponse::Error {
code: "rescan_cancelled".to_string(),
message: "daemon draining; partial scan committed".to_string(),
},
Ok(Ok((stats, false))) => CommsResponse::Rescanned {
scanned: stats.scanned,
updated: stats.updated,
docs_indexed: stats.docs_indexed,
removed: stats.removed,
elapsed_ms: started.elapsed().as_millis() as u64,
},
Ok(Err(error)) => CommsResponse::Error {
code: "rescan_failed".to_string(),
message: error.to_string(),
},
Err(join) => CommsResponse::Error {
code: "rescan_panicked".to_string(),
message: join.to_string(),
},
}
}
/// Report the daemon's currently-hot workspaces for the statusline.
pub(super) fn on_accessed_paths(&self) -> CommsResponse {
CommsResponse::Accessed {
workspaces: self.workspaces.accessed(),
}
}
/// List every registered workspace in the machine registry.
pub(super) async fn on_workspaces_list(&self) -> CommsResponse {
let registry = self.machine_registry.lock().await;
CommsResponse::Workspaces {
workspaces: registry.workspaces(),
}
}
/// List a registered repo's worktrees. An unknown repo id yields an empty list.
pub(super) async fn on_worktrees_list(&self, repo_id: String) -> CommsResponse {
let registry = self.machine_registry.lock().await;
CommsResponse::Worktrees {
worktrees: registry.worktrees(&repo_id),
}
}
/// List a registered repo's local branches. An unknown repo id yields an empty list.
pub(super) async fn on_branches_list(&self, repo_id: String) -> CommsResponse {
let registry = self.machine_registry.lock().await;
CommsResponse::Branches {
branches: registry.branches(&repo_id),
}
}
/// Advisory-claim a worktree. An unknown `(repo_id, name)` returns `held = false`.
pub(super) async fn on_worktree_claim(&self, repo_id: String, name: String, claimant: String) -> CommsResponse {
match self
.registry_blocking(move |registry| registry.claim_worktree(&repo_id, &name, &claimant))
.await
{
Ok(held) => CommsResponse::ClaimOutcome { held },
Err(error) => registry_error(error),
}
}
/// Drop machine-registry rows whose on-disk path is gone, returning how many were removed.
///
/// The registry is append-only by construction: a row is written when a workspace registers and
/// nothing ever retires it, so every throwaway checkout and test tempdir stays forever and buries
/// the live repos `workspace workspaces` exists to surface. Called from the daemon's periodic
/// maintenance pass; a failure is logged and reported as zero rather than propagated, because a
/// prune is opportunistic housekeeping and must never take the daemon down.
pub async fn prune_missing_registry_rows(&self) -> usize {
match self.registry_blocking(|registry| registry.prune_missing()).await {
Ok(removed) => removed,
Err(error) => {
tracing::warn!(%error, "comms: machine registry prune failed");
0
}
}
}
/// Release an advisory worktree claim held by `claimant`.
pub(super) async fn on_worktree_release(&self, repo_id: String, name: String, claimant: String) -> CommsResponse {
match self
.registry_blocking(move |registry| registry.release_worktree(&repo_id, &name, &claimant))
.await
{
Ok(held) => CommsResponse::ClaimOutcome { held },
Err(error) => registry_error(error),
}
}
}