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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
//! The daemon's git-history half: the pool of open per-repo history indexes, and the handler for the
//! forwarded [`GitHistoryOp`]s.
//!
//! Fjall's directory lock is exclusive — even a read-only open takes it — so `git-history.fjall/` has
//! exactly ONE holder. Under the daemon-as-sole-writer model that holder is the daemon: it BUILDS
//! each repo's index (the expensive history walk) and answers the front-ends' history reads from it,
//! exactly as it already owns the code index behind
//! [`WorkspacePool`](super::workspace_pool::WorkspacePool). A `daemon_writer` serve holds no handle
//! and forwards both halves here.
//!
//! Split out of `daemon.rs` to keep that file within the module-size cap.
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::Mutex;
use super::daemon::Broker;
use super::protocol::CommsResponse;
use crate::git_history::proto::{GitHistoryOp, GitHistoryReply};
use crate::git_history::{GitHistoryError, GitHistoryIndex};
/// One repo's open git-history index, held by the daemon (the only process allowed to hold it).
pub(crate) struct HistoryEntry {
/// The open fjall database. Shared by every link; fjall's keyspaces are thread-safe, so reads
/// run concurrently with each other and with an in-flight build.
index: Arc<GitHistoryIndex>,
/// Serializes SYNCS of this repo. This is the point of the daemon owning the build: N serve
/// sessions asking at once collapse to ONE history walk, and the rest observe `Fresh`.
build_lock: Arc<Mutex<()>>,
/// Drives the idle sweep, which drops the handle — and with it fjall's lock — for a cold repo.
last_used: std::sync::Mutex<Instant>,
}
impl HistoryEntry {
fn touch(&self) {
*self.last_used.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Instant::now();
}
/// How long this entry has sat unrequested. The daemon's periodic sweep sheds it past the TTL.
pub(crate) fn idle_for(&self) -> Duration {
self.last_used
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.elapsed()
}
}
impl Broker {
/// Run a forwarded git-history op against the repo's index — the BUILD and the reads, because the
/// daemon is the only process that may hold the database.
///
/// A [`GitHistoryOp::Sync`] takes the repo's build lock, so N sessions asking at once produce ONE
/// build: the winner walks history; the rest wait, then see `Fresh` (`builder::sync` compares
/// `last_indexed_head` to HEAD and no-ops). The walk is heavy and blocking, so it — and every
/// read — runs on a blocking thread while the reactor keeps serving other links.
pub(crate) async fn on_git_history(&self, root: std::path::PathBuf, op: GitHistoryOp) -> CommsResponse {
match self.run_git_history_inproc(root, op).await {
Ok(reply) => CommsResponse::GitHistory(reply),
Err(GitHistoryError::Disabled) => CommsResponse::Error {
code: "git_history_disabled".to_string(),
message: GitHistoryError::Disabled.to_string(),
},
Err(error) => CommsResponse::Error {
code: "git_history_failed".to_string(),
message: error.to_string(),
},
}
}
/// Run one git-history op against the repo's daemon-held index, in-process.
///
/// The single funnel for BOTH the forwarded socket path ([`on_git_history`](Self::on_git_history))
/// and a daemon-hosted connection (via [`HistoryHost`](crate::git_history::remote::HistoryHost)):
/// it opens (or reuses) the repo's one handle, serializes a [`Sync`](GitHistoryOp::Sync) on that
/// repo's build lock — so N callers arriving at once produce ONE history walk — and runs the op on
/// a blocking thread. Routing every caller here is what keeps the index single-holder and the
/// build single-flight no matter how the request arrived.
pub(crate) async fn run_git_history_inproc(
&self,
root: std::path::PathBuf,
op: GitHistoryOp,
) -> Result<GitHistoryReply, GitHistoryError> {
self.mark_active().await;
if !crate::git_history::index_enabled() {
return Err(GitHistoryError::Disabled);
}
let dir = crate::git_history::shared_history_basemind_dir(&root);
let entry = self.history_entry(&dir).await?;
entry.touch();
let _build_guard = match op {
GitHistoryOp::Sync => Some(entry.build_lock.clone().lock_owned().await),
_ => None,
};
// ~keep A first-build `Sync` on a deep repo runs for minutes in `spawn_blocking`, and the hosted
// ~keep path detaches it with no client link to keep the daemon busy. Pin it as in-flight work so
// ~keep the idle reaper cannot tear the process down mid-write to `git-history.fjall/` — the same
// ~keep guard `run_blob_gc_with_lock_timeout` takes for the other long client-less task.
let _working = matches!(op, GitHistoryOp::Sync).then(|| self.begin_work());
let index = Arc::clone(&entry.index);
let err_dir = dir.clone();
tokio::task::spawn_blocking(move || run_git_history_op(&index, &root, &dir, op))
.await
.map_err(|join| GitHistoryError::Io {
path: err_dir,
source: std::io::Error::other(join.to_string()),
})?
}
/// Fetch (or lazily open) the git-history index for `dir`.
///
/// The open takes fjall's exclusive lock, so it is serialized on `git_history_open_lock`: two
/// racing first-touches must not both call `open` on the same database — the loser would fail on
/// the lock instead of sharing the winner's handle (the same cold-open race the workspace pool
/// hit). Re-checks the map under the lock, so the loser returns the winner's entry.
async fn history_entry(&self, dir: &std::path::Path) -> Result<Arc<HistoryEntry>, GitHistoryError> {
if let Some(entry) = self.history_lookup(dir) {
return Ok(entry);
}
let _opening = self.git_history_open_lock.lock().await;
if let Some(entry) = self.history_lookup(dir) {
return Ok(entry);
}
let open_dir = dir.to_path_buf();
let opened = tokio::task::spawn_blocking(move || GitHistoryIndex::open(&open_dir))
.await
.map_err(|join| GitHistoryError::Io {
path: dir.to_path_buf(),
source: std::io::Error::other(join.to_string()),
})??;
let entry = Arc::new(HistoryEntry {
index: Arc::new(opened),
build_lock: Arc::new(Mutex::new(())),
last_used: std::sync::Mutex::new(Instant::now()),
});
self.git_history
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert(dir.to_path_buf(), Arc::clone(&entry));
Ok(entry)
}
fn history_lookup(&self, dir: &std::path::Path) -> Option<Arc<HistoryEntry>> {
self.git_history
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.get(dir)
.map(Arc::clone)
}
}
/// The in-process git-history seam: a daemon-hosted connection runs its history ops directly against
/// the daemon that hosts it, sharing the same single handle and build lock the forwarded path uses,
/// instead of dialing the daemon over its own socket.
impl crate::git_history::remote::HistoryHost for Broker {
fn run_history(
&self,
root: std::path::PathBuf,
op: GitHistoryOp,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<GitHistoryReply, GitHistoryError>> + Send + '_>>
{
Box::pin(self.run_git_history_inproc(root, op))
}
}
/// Execute one git-history op against the daemon-held index. Blocking (fjall reads, and a full
/// history walk for `Sync`); the caller runs it on a blocking thread.
fn run_git_history_op(
index: &GitHistoryIndex,
root: &std::path::Path,
dir: &std::path::Path,
op: GitHistoryOp,
) -> Result<GitHistoryReply, GitHistoryError> {
match op {
GitHistoryOp::Sync => {
let repo = crate::git::Repo::discover(root)?;
let outcome = crate::git_history::builder::sync(index, &repo, dir)?;
tracing::info!(root = %root.display(), ?outcome, "git-history index sync complete");
Ok(GitHistoryReply::Synced(outcome.into()))
}
GitHistoryOp::IndexedHead => Ok(GitHistoryReply::IndexedHead(index.last_indexed_head_hex())),
GitHistoryOp::RecentCommits {
skip,
take,
include_files,
} => Ok(GitHistoryReply::Commits(index.recent_commits(
skip,
take,
include_files,
))),
GitHistoryOp::CommitsTouching { path, skip, take } => {
Ok(GitHistoryReply::Commits(index.commits_touching(&path, skip, take)))
}
GitHistoryOp::WindowCommits { window } => Ok(GitHistoryReply::Commits(index.window_commits(window))),
GitHistoryOp::SearchCommits {
query,
scope,
skip,
take,
} => Ok(GitHistoryReply::Commits(
index.search_commits(&query, scope, skip, take),
)),
}
}