flagon-io/g1t

public

Merge last commits over the whole history: progress kept per ref and path, a push reads only its own commits

syntaqxcommitted 14h agoParentse42aa43 874677aBrowse files

5	5		//! tree hash differs), so the walk reads a tree only for the commits that
6	6		//! touched it. An entry is given the newest commit after which its hash is
7	7		//! no longer the same; one that never changes within the walk is given the
8		−	//! oldest commit reached if that is the root, and nothing otherwise.
	8	+	//! first commit there is, once the walk reaches it.
	9	+	//!
	10	+	//! A walk is never thrown away. What it found is remembered as
	11	+	//! [`Progress`] per repository, ref and path ([`Memo`]), with the head it
	12	+	//! started from and, when it had to stop, the commit to go on from:
	13	+	//!
	14	+	//! - The same head again goes on from where the last walk stopped, so a
	15	+	//! long history is walked to its first commit over as many calls as it
	16	+	//! takes, each within the Worker's limits ([`MAX_READS`]).
	17	+	//! - A new head walks only back to the remembered one, when that is on its
	18	+	//! first-parent chain: an entry whose hash did not change in between keeps
	19	+	//! the commit remembered for it. A push then costs only its own commits.
	20	+	//! When the remembered head never comes up (a force-push), the walk is a
	21	+	//! full one.
9	22
10		−	use std::collections::HashMap;
	23	+	use std::collections::{HashMap, HashSet};
11	24
12	25		use g1t_contracts::repos::{Commit, EntryKind, LastCommit, TreeEntry};
	26	+	use serde::{Deserialize, Serialize};
13	27		use worker::Result;
14	28
	29	+	use crate::shared::Shared;
15	30		use crate::store::GitRepo;
16	31
17		−	/// How far back the history is walked.
18		−	pub const MAX_COMMITS: u32 = 300;
	32	+	/// Reads of the store one call makes at most. Each costs up to three
	33	+	/// subrequests (a Cache API look, the store, a Cache API put), so this
	34	+	/// keeps a call well inside the 10,000 a Worker invocation may make, with
	35	+	/// room for the rest of the request. A repository's root takes about one
	36	+	/// read per commit, so a call walks some 2,000 commits, and the next call
	37	+	/// goes on from there.
	38	+	pub const MAX_READS: u32 = 2_500;
19	39		/// Commits whose trees are read together, ahead of the walk: each read is a
20	40		/// round trip to the store, so reading them one by one is what is slow.
21	41		const READ_AHEAD: usize = 24;
22	42		/// Commits of history read at a time.
23	43		const PAGE: u32 = 48;
24	44
	45	+	/// What walks found for a directory, from `head`: kept between walks.
	46	+	#[derive(Clone, Debug, Serialize, Deserialize)]
	47	+	pub struct Progress {
	48	+	/// The commit the walks started from.
	49	+	pub head: String,
	50	+	/// The entries given their commit.
	51	+	pub found: Vec<LastCommit>,
	52	+	/// The entries not given one yet.
	53	+	pub open: Vec<String>,
	54	+	/// The commit to go on from while some are open: the newest not yet
	55	+	/// compared with its parent. `None` once no walk can find more.
	56	+	pub at: Option<String>,
	57	+	}
	58	+
	59	+	impl Progress {
	60	+	pub fn complete(&self) -> bool {
	61	+	self.open.is_empty()
	62	+	}
	63	+
	64	+	/// Whether no walk could find more.
	65	+	pub fn settled(&self) -> bool {
	66	+	self.open.is_empty() \|\| self.at.is_none()
	67	+	}
	68	+	}
	69	+
	70	+	/// One call's walk: where it got to, and how many reads of the store it made.
	71	+	pub struct Walk {
	72	+	pub progress: Progress,
	73	+	pub reads: u32,
	74	+	}
	75	+
25	76		/// Reads trees, remembering those already read: commits share most of them.
26	77		struct Trees<'a, R: GitRepo> {
27	78		repo: &'a R,
28	79		read: HashMap<String, Option<Vec<TreeEntry>>>,
	80	+	/// Reads of the store so far, trees and history.
	81	+	reads: u32,
29	82		}
30	83
31	84		impl<'a, R: GitRepo> Trees<'a, R> {
	85	+	async fn log(&mut self, from: &str) -> Result<Vec<Commit>> {
	86	+	self.reads += 1;
	87	+	self.repo.log(from, PAGE).await
	88	+	}
	89	+
32	90		/// Reads the trees not read yet, all at once.
33	91		async fn prefetch(&mut self, hashes: impl IntoIterator<Item = String>) -> Result<()> {
34	92		let mut wanted: Vec<String> = hashes.into_iter().filter(\|hash\| !self.read.contains_key(hash)).collect();
35	93		wanted.sort();
36	94		wanted.dedup();
	95	+	self.reads += wanted.len() as u32;
37	96		let found = futures_util::future::join_all(wanted.iter().map(\|hash\| self.repo.read_tree(hash))).await;
38	97		for (hash, tree) in wanted.into_iter().zip(found) {
39	98		self.read.insert(hash, tree?);
⋯
61	120		if let Some(found) = self.read.get(hash) {
62	121		return Ok(found.clone());
63	122		}
	123	+	self.reads += 1;
64	124		let found = self.repo.read_tree(hash).await?;
65	125		self.read.insert(hash.to_owned(), found.clone());
66	126		Ok(found)
⋯
89	149		}
90	150		}
91	151
92		−	/// The last commit of each entry of `path` at `git_ref`, and whether every
93		−	/// entry was given one. `out_of_time` is asked between batches; once it
94		−	/// says so the walk stops with what it has, incomplete.
	152	+	/// Where a walk stands: the history ahead from the commit it is at, and
	153	+	/// that commit's directory and entries.
	154	+	struct Cursor {
	155	+	history: Vec<Commit>,
	156	+	index: usize,
	157	+	/// The trees of history before this index have been read ahead.
	158	+	read_ahead_to: usize,
	159	+	dir: Option<String>,
	160	+	current: HashMap<String, String>,
	161	+	}
	162	+
	163	+	impl Cursor {
	164	+	/// At the commit `from`, or `None` when there is no such commit.
	165	+	async fn at<R: GitRepo>(trees: &mut Trees<'_, R>, from: &str, path: &str) -> Result<Option<Cursor>> {
	166	+	let history = trees.log(from).await?;
	167	+	let Some(first) = history.first() else {
	168	+	return Ok(None);
	169	+	};
	170	+	let dir = trees.dir(&first.tree_hash.clone(), path).await?;
	171	+	let current = trees.entries(dir.as_deref()).await?;
	172	+	Ok(Some(Cursor { history, index: 0, read_ahead_to: 0, dir, current }))
	173	+	}
	174	+
	175	+	fn commit(&self) -> &Commit {
	176	+	&self.history[self.index]
	177	+	}
	178	+	}
	179	+
	180	+	/// The last commit of each entry of `path` at the commit `head`, going on
	181	+	/// from `kept` (what earlier walks of the same ref and path found). The
	182	+	/// walk stops, for the next call to go on from, once `out_of_time` says so
	183	+	/// (asked between batches, after some progress) or before it would make
	184	+	/// more than `max_reads` reads of the store.
95	185		pub async fn last_commits<R: GitRepo>(
96	186		repo: &R,
97		−	git_ref: &str,
	187	+	head: &str,
98	188		path: &str,
	189	+	kept: Option<Progress>,
99	190		out_of_time: &dyn Fn() -> bool,
100		−	) -> Result<(Vec<LastCommit>, bool)> {
101		−	// History a page at a time, so a walk that is out of time stops
102		−	// between pages rather than after reading all of it.
103		−	let mut history = repo.log(git_ref, PAGE).await?;
104		−	let Some(head) = history.first().cloned() else {
105		−	return Ok((Vec::new(), true));
106		−	};
107		−	let mut trees = Trees { repo, read: HashMap::new() };
108		−	let mut dir = trees.dir(&head.tree_hash, path).await?;
109		−	let mut current = trees.entries(dir.as_deref()).await?;
110		−	let mut open: Vec<String> = current.keys().cloned().collect();
	191	+	max_reads: u32,
	192	+	) -> Result<Walk> {
	193	+	let mut trees = Trees { repo, read: HashMap::new(), reads: 0 };
	194	+	let segments = path.split('/').filter(\|segment\| !segment.is_empty()).count();
	195	+	// The most a batch read ahead reads, and a page of history.
	196	+	let batch = (READ_AHEAD * (segments + 1)) as u32 + 1;
111	197		let mut found: Vec<LastCommit> = Vec::new();
	198	+	let mut open: Vec<String> = Vec::new();
	199	+	// An earlier walk from another head, to stop at if it comes up.
	200	+	let mut stop_at: Option<Progress> = None;
	201	+	let fresh;
	202	+	let start = match kept {
	203	+	Some(kept) if kept.head == head => {
	204	+	let Some(at) = kept.at.clone().filter(\|_\| !kept.open.is_empty()) else {
	205	+	return Ok(Walk { progress: kept, reads: 0 });
	206	+	};
	207	+	found = kept.found;
	208	+	open = kept.open;
	209	+	fresh = false;
	210	+	at
	211	+	}
	212	+	other => {
	213	+	stop_at = other;
	214	+	fresh = true;
	215	+	head.to_owned()
	216	+	}
	217	+	};
	218	+	let Some(mut cursor) = Cursor::at(&mut trees, &start, path).await? else {
	219	+	// An unknown head has nothing. A commit to go on from that cannot
	220	+	// be read any more leaves what is open unknown.
	221	+	return Ok(Walk { progress: Progress { head: head.to_owned(), found, open, at: None }, reads: trees.reads });
	222	+	};
	223	+	if fresh {
	224	+	open = cursor.current.keys().cloned().collect();
	225	+	}
112	226		let give = \|found: &mut Vec<LastCommit>, name: String, commit: &Commit\| found.push(LastCommit { name, commit: commit.clone() });
113		−	let mut index = 0;
114		−	while !open.is_empty() && index < history.len() {
115		−	// Read the next page once the walk reaches the end of this one.
116		−	if index + 1 >= history.len() && history.len() < MAX_COMMITS as usize && !out_of_time() {
117		−	if let Some(parent) = history.last().and_then(\|commit\| commit.parents.first()).cloned() {
118		−	let more = repo.log(&parent, PAGE).await?;
119		−	history.extend(more);
	227	+	let mut walked = 0u32;
	228	+	// Whether the walk had to stop with more to find.
	229	+	let mut stopped = false;
	230	+	while !open.is_empty() {
	231	+	// The head an earlier walk started from: what has not changed
	232	+	// since keeps what that walk found for it.
	233	+	if stop_at.as_ref().is_some_and(\|kept\| kept.head == cursor.commit().hash) {
	234	+	let Some(kept) = stop_at.take() else { break };
	235	+	let given: HashMap<&str, &LastCommit> = kept.found.iter().map(\|last\| (last.name.as_str(), last)).collect();
	236	+	let kept_open: HashSet<&str> = kept.open.iter().map(String::as_str).collect();
	237	+	let mut rest = Vec::new();
	238	+	for name in open.drain(..) {
	239	+	match given.get(name.as_str()) {
	240	+	Some(last) => found.push((*last).clone()),
	241	+	None => rest.push(name),
	242	+	}
	243	+	}
	244	+	open = rest;
	245	+	// The rest were still open for that walk too: go on from where
	246	+	// it stopped rather than walk the same history again. Otherwise
	247	+	// (it did not know them) the walk goes on from here.
	248	+	if !open.is_empty() && open.iter().all(\|name\| kept_open.contains(name.as_str())) {
	249	+	let next = match &kept.at {
	250	+	Some(at) => Cursor::at(&mut trees, at, path).await?,
	251	+	None => None,
	252	+	};
	253	+	// None: that walk found all there was to find.
	254	+	let Some(next) = next else { break };
	255	+	cursor = next;
120	256		}
	257	+	continue;
	258	+	}
	259	+	let needs_page = cursor.index + 1 >= cursor.history.len() && !cursor.commit().parents.is_empty();
	260	+	let needs_read_ahead = cursor.index >= cursor.read_ahead_to;
	261	+	if (needs_page \|\| needs_read_ahead) && walked > 0 && (out_of_time() \|\| trees.reads + batch > max_reads) {
	262	+	stopped = true;
	263	+	break;
121	264		}
122		−	if index % READ_AHEAD == 0 {
123		−	if index > 0 && out_of_time() {
124		−	break;
125		−	}
126		−	let ahead = history.iter().skip(index + 1).take(READ_AHEAD).map(\|commit\| commit.tree_hash.clone()).collect();
	265	+	// The next page once the walk reaches the end of this one.
	266	+	if needs_page {
	267	+	let parent = cursor.commit().parents[0].clone();
	268	+	let more = trees.log(&parent).await?;
	269	+	// What is behind the walk is not needed again.
	270	+	cursor.history.drain(..cursor.index);
	271	+	cursor.read_ahead_to = cursor.read_ahead_to.saturating_sub(cursor.index);
	272	+	cursor.index = 0;
	273	+	cursor.history.extend(more);
	274	+	}
	275	+	if cursor.index >= cursor.read_ahead_to {
	276	+	// Nor are the trees already compared.
	277	+	trees.read.clear();
	278	+	let ahead = cursor.history.iter().skip(cursor.index + 1).take(READ_AHEAD).map(\|commit\| commit.tree_hash.clone()).collect();
127	279		trees.prefetch_dirs(ahead, path).await?;
	280	+	cursor.read_ahead_to = cursor.index + READ_AHEAD;
128	281		}
129		−	let commit = history[index].clone();
130		−	let Some(parent) = history.get(index + 1).cloned() else {
131		−	// The oldest commit read. If it is the first commit there is,
132		−	// what is left was added by it.
	282	+	let commit = cursor.commit().clone();
	283	+	let Some(parent) = cursor.history.get(cursor.index + 1).cloned() else {
	284	+	// The oldest commit there is. If it is the first commit, what
	285	+	// is left was added by it; if its parent could not be read,
	286	+	// what is left is not known.
133	287		if commit.parents.is_empty() {
134	288		for name in open.drain(..) {
135	289		give(&mut found, name, &commit);
⋯
137	291		}
138	292		break;
139	293		};
140		−	index += 1;
	294	+	cursor.index += 1;
	295	+	walked += 1;
141	296		let parent_dir = trees.dir(&parent.tree_hash, path).await?;
142		−	if parent_dir == dir {
	297	+	if parent_dir == cursor.dir {
143	298		continue;
144	299		}
145	300		let before = trees.entries(parent_dir.as_deref()).await?;
146		−	let (changed, still): (Vec<String>, Vec<String>) = open.into_iter().partition(\|name\| before.get(name) != current.get(name));
	301	+	let (changed, still): (Vec<String>, Vec<String>) = open.into_iter().partition(\|name\| before.get(name) != cursor.current.get(name));
147	302		for name in changed {
148	303		give(&mut found, name, &commit);
149	304		}
150	305		open = still;
151		−	dir = parent_dir;
152		−	current = before;
	306	+	cursor.dir = parent_dir;
	307	+	cursor.current = before;
	308	+	}
	309	+	let at = (stopped && !open.is_empty()).then(\|\| cursor.commit().hash.clone());
	310	+	Ok(Walk { progress: Progress { head: head.to_owned(), found, open, at }, reads: trees.reads })
	311	+	}
	312	+
	313	+	/// Where the progress for a ref and path of a repository is kept: in this
	314	+	/// colo's cache, and shared between colos when the service has `GIT_CACHE`
	315	+	/// (shared.rs).
	316	+	pub struct Memo {
	317	+	colo_url: String,
	318	+	shared_key: String,
	319	+	}
	320	+
	321	+	/// How long progress is kept after its last walk.
	322	+	const MEMO_TTL_SECONDS: u64 = 30 * 24 * 60 * 60;
	323	+
	324	+	impl Memo {
	325	+	pub fn new(repo_id: &str, git_ref: &str, path: &str) -> Memo {
	326	+	let what = g1t_secrets::sha256_hex(&format!("{git_ref}\n{path}"));
	327	+	Memo {
	328	+	colo_url: format!("https://last-commits.g1t.internal/progress/v1/{repo_id}/{what}"),
	329	+	shared_key: format!("last-commits:{repo_id}:{what}"),
	330	+	}
	331	+	}
	332	+
	333	+	/// The progress kept, this colo's first. A failure to read is none.
	334	+	pub async fn get(&self, shared: Option<&Shared>) -> Option<Progress> {
	335	+	if let Ok(Some(mut response)) = worker::Cache::default().get(self.colo_url.as_str(), false).await
	336	+	&& let Ok(progress) = response.json::<Progress>().await
	337	+	{
	338	+	return Some(progress);
	339	+	}
	340	+	let bytes = shared?.get(&self.shared_key).await?;
	341	+	serde_json::from_slice(&bytes).ok()
	342	+	}
	343	+
	344	+	/// Keeps `progress` here and in the shared store. A failure only costs
	345	+	/// a longer walk later.
	346	+	pub async fn keep(&self, shared: Option<&Shared>, progress: &Progress) {
	347	+	let Ok(bytes) = serde_json::to_vec(progress) else {
	348	+	return;
	349	+	};
	350	+	let colo = async {
	351	+	if let Ok(mut response) = worker::Response::from_bytes(bytes.clone()) {
	352	+	let _ = response.headers_mut().set("cache-control", &format!("max-age={MEMO_TTL_SECONDS}"));
	353	+	let _ = worker::Cache::default().put(self.colo_url.as_str(), response).await;
	354	+	}
	355	+	};
	356	+	let shared_put = async {
	357	+	if let Some(shared) = shared {
	358	+	shared.put(&self.shared_key, &bytes, MEMO_TTL_SECONDS).await;
	359	+	}
	360	+	};
	361	+	futures_util::future::join(colo, shared_put).await;
153	362		}
154		−	let complete = open.is_empty();
155		−	Ok((found, complete))
156	363		}
157	364
158	365		#[cfg(test)]
159	366		mod tests {
	367	+	use std::cell::RefCell;
160	368		use std::future::Future;
161	369		use std::pin::pin;
162	370		use std::task::{Context, Poll, Waker};
⋯
177	385		struct Fake {
178	386		trees: HashMap<String, Vec<TreeEntry>>,
179	387		history: Vec<Commit>,
	388	+	/// Every tree read, in order.
	389	+	trees_read: RefCell<Vec<String>>,
	390	+	/// Where each history read started.
	391	+	logs_read: RefCell<Vec<String>>,
180	392		}
181	393
182	394		impl GitRepo for Fake {
⋯
187	399		Ok(Vec::new())
188	400		}
189	401		async fn log(&self, git_ref: &str, limit: u32) -> Result<Vec<Commit>> {
	402	+	self.logs_read.borrow_mut().push(git_ref.to_owned());
190	403		// A branch name starts at the head; a hash at that commit; anything else is unknown.
191	404		let start = if git_ref == "main" { Some(0) } else { self.history.iter().position(\|commit\| commit.hash == git_ref) };
192	405		Ok(start.map(\|start\| self.history.iter().skip(start).take(limit as usize).cloned().collect()).unwrap_or_default())
⋯
195	408		Ok(None)
196	409		}
197	410		async fn read_tree(&self, tree_hash: &str) -> Result<Option<Vec<TreeEntry>>> {
	411	+	self.trees_read.borrow_mut().push(tree_hash.to_owned());
198	412		Ok(self.trees.get(tree_hash).cloned())
199	413		}
200	414		async fn read_blob(&self, _blob_hash: &str) -> Result<Option<Vec<u8>>> {
⋯
235	449		fake
236	450		}
237	451
238		−	fn by_name(found: Vec<LastCommit>) -> HashMap<String, String> {
239		−	found.into_iter().map(\|last\| (last.name, last.commit.hash)).collect()
	452	+	/// A long history, newest first: `.dockerignore` is added by the first
	453	+	/// commit and never changed; `README.md` changes in every commit after
	454	+	/// it, under a tree of its own (`t<n>`).
	455	+	fn long(commits: usize) -> Fake {
	456	+	let mut fake = Fake::default();
	457	+	for n in 0..commits {
	458	+	let tree = vec![entry(".dockerignore", "ignore", EntryKind::Blob), entry("README.md", &format!("r{n}"), EntryKind::Blob)];
	459	+	fake.trees.insert(format!("t{n}"), tree);
	460	+	let parent = (n > 0).then(\|\| format!("c{}", n - 1));
	461	+	fake.history.insert(0, commit(&format!("c{n}"), &format!("t{n}"), parent.as_deref()));
	462	+	}
	463	+	fake
240	464		}
241	465
	466	+	/// Adds commits on top: each changes README.md; `touch_new` also adds `NEW`.
	467	+	fn push(fake: &mut Fake, commits: usize, touch_new: bool) {
	468	+	let first = fake.history.len();
	469	+	for n in first..first + commits {
	470	+	let mut tree = vec![entry(".dockerignore", "ignore", EntryKind::Blob), entry("README.md", &format!("r{n}"), EntryKind::Blob)];
	471	+	if touch_new {
	472	+	tree.push(entry("NEW", "new", EntryKind::Blob));
	473	+	}
	474	+	fake.trees.insert(format!("t{n}"), tree);
	475	+	fake.history.insert(0, commit(&format!("c{n}"), &format!("t{n}"), Some(&format!("c{}", n - 1))));
	476	+	}
	477	+	}
	478	+
	479	+	fn walk(fake: &Fake, head: &str, path: &str, kept: Option<Progress>) -> Progress {
	480	+	run(last_commits(fake, head, path, kept, &\|\| false, MAX_READS)).unwrap().progress
	481	+	}
	482	+
	483	+	fn by_name(found: &[LastCommit]) -> HashMap<String, String> {
	484	+	found.iter().map(\|last\| (last.name.clone(), last.commit.hash.clone())).collect()
	485	+	}
	486	+
242	487		#[test]
243	488		fn each_root_entry_gets_the_newest_commit_that_changed_it() {
244		−	let (found, complete) = run(last_commits(&repo(), "main", "", &\|\| false)).unwrap();
245		−	assert!(complete);
246		−	let found = by_name(found);
	489	+	let found = walk(&repo(), "c3", "", None);
	490	+	assert!(found.complete());
	491	+	let found = by_name(&found.found);
247	492		assert_eq!(found["README.md"], "c3");
248	493		assert_eq!(found["src"], "c2");
249	494		}
250	495
251	496		#[test]
252	497		fn a_subdirectory_is_walked_by_its_own_tree() {
253		−	let (found, complete) = run(last_commits(&repo(), "main", "src", &\|\| false)).unwrap();
254		−	assert!(complete);
255		−	assert_eq!(by_name(found)["a.rs"], "c2");
	498	+	let found = walk(&repo(), "c3", "src", None);
	499	+	assert!(found.complete());
	500	+	assert_eq!(by_name(&found.found)["a.rs"], "c2");
256	501		}
257	502
258	503		#[test]
⋯
260	505		let mut fake = repo();
261	506		fake.trees.insert("root2".into(), vec![entry("README.md", "r1", EntryKind::Blob), entry("src", "src1", EntryKind::Tree)]);
262	507		fake.trees.insert("root3".into(), vec![entry("README.md", "r2", EntryKind::Blob), entry("src", "src1", EntryKind::Tree)]);
263		−	let (found, complete) = run(last_commits(&fake, "main", "", &\|\| false)).unwrap();
264		−	assert!(complete);
265		−	assert_eq!(by_name(found)["src"], "c1");
	508	+	let found = walk(&fake, "c3", "", None);
	509	+	assert!(found.complete());
	510	+	assert_eq!(by_name(&found.found)["src"], "c1");
266	511		}
267	512
268	513		#[test]
269		−	fn a_walk_cut_short_leaves_the_rest_unknown() {
	514	+	fn history_that_cannot_be_read_leaves_the_rest_unknown_for_good() {
270	515		let mut fake = repo();
271	516		// c1 has a parent the walk never reaches.
272	517		fake.history[2].parents = vec!["c0".into()];
273	518		fake.trees.insert("root2".into(), vec![entry("README.md", "r1", EntryKind::Blob), entry("src", "src1", EntryKind::Tree)]);
274	519		fake.trees.insert("root3".into(), vec![entry("README.md", "r2", EntryKind::Blob), entry("src", "src1", EntryKind::Tree)]);
275		−	let (found, complete) = run(last_commits(&fake, "main", "", &\|\| false)).unwrap();
276		−	assert!(!complete);
277		−	let found = by_name(found);
278		−	assert_eq!(found["README.md"], "c3");
279		−	assert!(!found.contains_key("src"));
	520	+	let found = walk(&fake, "c3", "", None);
	521	+	assert!(!found.complete());
	522	+	assert!(found.settled(), "no later walk can find more");
	523	+	let names = by_name(&found.found);
	524	+	assert_eq!(names["README.md"], "c3");
	525	+	assert!(!names.contains_key("src"));
	526	+	}
	527	+
	528	+	#[test]
	529	+	fn an_entry_older_than_a_thousand_commits_resolves() {
	530	+	let fake = long(1_200);
	531	+	let found = walk(&fake, "c1199", "", None);
	532	+	assert!(found.complete());
	533	+	let names = by_name(&found.found);
	534	+	assert_eq!(names[".dockerignore"], "c0");
	535	+	assert_eq!(names["README.md"], "c1199");
	536	+	}
	537	+
	538	+	#[test]
	539	+	fn a_walk_out_of_reads_stops_and_the_next_goes_on_from_there() {
	540	+	let fake = long(1_000);
	541	+	let first = run(last_commits(&fake, "c999", "", None, &\|\| false, 300)).unwrap();
	542	+	assert!(first.reads <= 300, "{} reads", first.reads);
	543	+	assert!(!first.progress.settled());
	544	+	let at = first.progress.at.clone().unwrap();
	545	+	assert_ne!(at, "c999");
	546	+	let mut progress = first.progress;
	547	+	let mut calls = 1;
	548	+	while !progress.settled() {
	549	+	fake.trees_read.borrow_mut().clear();
	550	+	let next = run(last_commits(&fake, "c999", "", Some(progress), &\|\| false, 300)).unwrap();
	551	+	assert!(next.reads <= 300);
	552	+	// The next call starts where the last one stopped.
	553	+	assert!(!fake.trees_read.borrow().contains(&"t999".to_owned()));
	554	+	progress = next.progress;
	555	+	calls += 1;
	556	+	}
	557	+	assert!(calls >= 3);
	558	+	assert!(progress.complete());
	559	+	assert_eq!(by_name(&progress.found)[".dockerignore"], "c0");
	560	+	}
	561	+
	562	+	#[test]
	563	+	fn three_thousand_commits_take_two_calls_within_the_read_limit() {
	564	+	let fake = long(3_000);
	565	+	let first = run(last_commits(&fake, "c2999", "", None, &\|\| false, MAX_READS)).unwrap();
	566	+	assert!(first.reads <= MAX_READS);
	567	+	assert!(!first.progress.settled());
	568	+	let second = run(last_commits(&fake, "c2999", "", Some(first.progress), &\|\| false, MAX_READS)).unwrap();
	569	+	assert!(second.reads <= MAX_READS);
	570	+	assert!(second.progress.complete());
	571	+	assert_eq!(by_name(&second.progress.found)[".dockerignore"], "c0");
	572	+	}
	573	+
	574	+	#[test]
	575	+	fn out_of_time_stops_after_some_progress() {
	576	+	let fake = long(500);
	577	+	let found = run(last_commits(&fake, "c499", "", None, &\|\| true, MAX_READS)).unwrap().progress;
	578	+	assert!(!found.settled());
	579	+	assert_eq!(by_name(&found.found)["README.md"], "c499");
	580	+	}
	581	+
	582	+	#[test]
	583	+	fn a_new_head_reads_only_the_commits_since_the_remembered_one() {
	584	+	let mut fake = long(600);
	585	+	let before = walk(&fake, "c599", "", None);
	586	+	assert!(before.complete());
	587	+	push(&mut fake, 3, true);
	588	+	fake.trees_read.borrow_mut().clear();
	589	+	fake.logs_read.borrow_mut().clear();
	590	+	let after = walk(&fake, "c602", "", Some(before));
	591	+	assert!(after.complete());
	592	+	let names = by_name(&after.found);
	593	+	assert_eq!(names[".dockerignore"], "c0", "kept from the earlier walk");
	594	+	assert_eq!(names["README.md"], "c602");
	595	+	assert_eq!(names["NEW"], "c600");
	596	+	// One page of history and one batch of trees read ahead, not 600.
	597	+	let read = fake.trees_read.borrow();
	598	+	assert!(read.len() <= READ_AHEAD + 1, "{read:?}");
	599	+	assert!(!read.contains(&"t500".to_owned()));
	600	+	assert_eq!(fake.logs_read.borrow().len(), 1);
	601	+	}
	602	+
	603	+	#[test]
	604	+	fn a_new_head_goes_on_from_where_an_unfinished_walk_stopped() {
	605	+	let mut fake = long(1_000);
	606	+	let first = run(last_commits(&fake, "c999", "", None, &\|\| false, 300)).unwrap().progress;
	607	+	assert!(!first.settled());
	608	+	let stopped_at = first.at.clone().unwrap();
	609	+	push(&mut fake, 2, false);
	610	+	fake.logs_read.borrow_mut().clear();
	611	+	let next = run(last_commits(&fake, "c1001", "", Some(first), &\|\| false, MAX_READS)).unwrap().progress;
	612	+	assert!(next.complete());
	613	+	assert_eq!(by_name(&next.found)[".dockerignore"], "c0");
	614	+	assert_eq!(by_name(&next.found)["README.md"], "c1001");
	615	+	// From the new head, then straight to where the first walk stopped.
	616	+	assert_eq!(fake.logs_read.borrow()[..2], ["c1001".to_owned(), stopped_at]);
	617	+	}
	618	+
	619	+	#[test]
	620	+	fn a_force_push_walks_the_whole_history_again() {
	621	+	let mut fake = long(400);
	622	+	let before = walk(&fake, "c399", "", None);
	623	+	assert!(before.complete());
	624	+	// A new history: .dockerignore is different in its first commit, and
	625	+	// the remembered head is not on it.
	626	+	let mut rewritten = long(450);
	627	+	for commit in &mut rewritten.history {
	628	+	commit.hash = format!("x{}", commit.hash);
	629	+	commit.parents = commit.parents.iter().map(\|parent\| format!("x{parent}")).collect();
	630	+	}
	631	+	rewritten.trees.get_mut("t0").unwrap()[0].hash = "old-ignore".into();
	632	+	for tree in (1..450).map(\|n\| format!("t{n}")) {
	633	+	rewritten.trees.get_mut(&tree).unwrap()[0].hash = "new-ignore".into();
	634	+	}
	635	+	fake = rewritten;
	636	+	let after = walk(&fake, "xc449", "", Some(before));
	637	+	assert!(after.complete());
	638	+	let names = by_name(&after.found);
	639	+	assert_eq!(names[".dockerignore"], "xc1", "found by walking, not kept from the old history");
	640	+	assert_eq!(names["README.md"], "xc449");
	641	+	assert!(fake.trees_read.borrow().contains(&"t1".to_owned()));
	642	+	}
	643	+
	644	+	#[test]
	645	+	fn the_same_head_again_reads_nothing() {
	646	+	let fake = repo();
	647	+	let before = walk(&fake, "c3", "", None);
	648	+	fake.trees_read.borrow_mut().clear();
	649	+	let again = run(last_commits(&fake, "c3", "", Some(before), &\|\| false, MAX_READS)).unwrap();
	650	+	assert_eq!(again.reads, 0);
	651	+	assert!(again.progress.complete());
280	652		}
281	653		}

77	77		const MAX_TAGS_READ: usize = 100;
78	78		/// Branch heads measured in one `branch_drift` call.
79	79		const MAX_DRIFT_HEADS: usize = 100;
	80	+	/// How long a `last_commits` walk asked without a budget runs before it
	81	+	/// stops and keeps its progress for the next call. The site asks from a
	82	+	/// waitUntil, which may run 30 s past its response.
	83	+	const LAST_COMMITS_WALK_MS: u64 = 20_000;
80	84
81	85		/// One path segment, percent-encoded for a cache key.
82	86		fn urlencoding_segment(segment: &str) -> String {
⋯
903	907		Ok(Outcome::Ok(git.log(&git_ref, a.limit).await?))
904	908		}
905	909
906		−	/// Which commit last changed each entry of a directory. Kept in this
907		−	/// colo's cache by repository, head commit and path: a commit's history
908		−	/// never changes, so an answer is good for as long as it is kept.
	910	+	/// Which commit last changed each entry of a directory. A finished
	911	+	/// answer is kept in this colo's cache by repository, head commit and
	912	+	/// path: a commit's history never changes, so it is good for as long as
	913	+	/// it is kept. Every walk also keeps its progress by ref and path
	914	+	/// (last_commits.rs), so the next call goes on from it: from where it
	915	+	/// stopped, or for a new head, only back to the old one.
909	916		async fn last_commits(&self, a: g1t_contracts::repos::LastCommitsArgs) -> Result<Outcome<g1t_contracts::repos::LastCommits>> {
910	917		let Some(repo) = self.readable(&a.path, &a.viewer).await? else {
911	918		return Ok(not_found());
⋯
915	922		let Some(head) = git.log(&git_ref, 1).await?.into_iter().next() else {
916	923		return Ok(Outcome::fail(FailureCode::NotFound, "No such branch, tag or commit."));
917	924		};
	925	+	// v2: v1 kept walks cut short at 300 commits as if finished.
918	926		let key = format!(
919		−	"https://last-commits.g1t.internal/{}/{}/{}",
	927	+	"https://last-commits.g1t.internal/v2/{}/{}/{}",
920	928		repo.id,
921	929		head.hash,
922	930		a.tree_path.split('/').map(urlencoding_segment).collect::<Vec<_>>().join("/")
⋯
927	935		return Ok(Outcome::Ok(found));
928	936		}
929	937		}
930		−	// Asked with a budget: past it, what was found so far, not kept.
	938	+	let memo = last_commits::Memo::new(&repo.id, &git_ref, &a.tree_path);
	939	+	let shared = self.shared.as_deref();
	940	+	let progress = memo.get(shared).await;
	941	+	// Asked with a budget, the walk stops past it; without one, past
	942	+	// LAST_COMMITS_WALK_MS, well before the caller's waitUntil ends.
931	943		let started = worker::Date::now().as_millis();
932		−	let budget = a.budget_ms;
933		−	let out_of_time = move \|\| budget.is_some_and(\|budget\| worker::Date::now().as_millis().saturating_sub(started) > budget);
934		−	let (entries, complete) = last_commits::last_commits(&git, &head.hash, &a.tree_path, &out_of_time).await?;
935		−	let stopped = out_of_time();
936		−	let found = g1t_contracts::repos::LastCommits { entries, complete };
937		−	if stopped && !found.complete {
	944	+	let budget = a.budget_ms.unwrap_or(LAST_COMMITS_WALK_MS);
	945	+	let out_of_time = move \|\| worker::Date::now().as_millis().saturating_sub(started) > budget;
	946	+	let walk = last_commits::last_commits(&git, &head.hash, &a.tree_path, progress, &out_of_time, last_commits::MAX_READS).await?;
	947	+	if walk.reads > 0 {
	948	+	memo.keep(shared, &walk.progress).await;
	949	+	}
	950	+	let settled = walk.progress.settled();
	951	+	let found = g1t_contracts::repos::LastCommits { complete: walk.progress.complete(), entries: walk.progress.found };
	952	+	if !settled {
938	953		return Ok(Outcome::Ok(found));
939	954		}
940	955		if let Ok(mut response) = worker::Response::from_json(&found) {

3	3		* answer not yet kept walks history, which can take seconds on a large or
4	4		* busy repository; the page goes out without it then, the column empty,
5	5		* while the walk finishes in the background (waitUntil) so the next view
6		−	* has it from the cache. The page's stream never waits on it past its own
7		−	* timeout.
	6	+	* has it from the cache. A history too long for one walk is read over
	7	+	* several views: the repos service keeps each walk's progress and goes on
	8	+	* from it (services/repos/src/last_commits.rs). The page's stream never
	9	+	* waits on it past its own timeout.
8	10		*/
9	11		import { waitUntil } from "cloudflare:workers";
10	12

142	142		the compared branch.
143	143
144	144		On Files, each file and folder shows the commit that last changed it and
145		−	when, from up to 300 commits of the branch's history; one changed before
146		−	that shows none. The branch menu at the top switches branch and keeps the
	145	+	when, from the branch's whole history. On a long history the first view
	146	+	can show some of them blank while g1t finishes reading it; a later view
	147	+	fills them in, and after a push only the new commits are read. The branch menu at the top switches branch and keeps the
147	148		folder or file you are on.
148	149
149	150		## Pull request forks

596	596		pub git_ref: Option<String>,
597	597		#[serde(default)]
598	598		pub tree_path: String,
599		−	/// Answer within this many milliseconds with what was found, not kept;
600		−	/// absent, the walk runs to the end and is kept.
	599	+	/// Answer within this many milliseconds with what was found; absent,
	600	+	/// within 20 seconds. A walk that stops short keeps its progress, and
	601	+	/// the next call goes on from it.
601	602		#[serde(default)]
602	603		pub budget_ms: Option<u64>,
603	604		}
⋯
609	610		pub commit: Commit,
610	611		}
611	612
612		−	/// The entries' last commits. `complete` is false when the history walked
613		−	/// ran out before every entry was placed; those entries are left out.
	613	+	/// The entries' last commits. `complete` is false when the walk stopped
	614	+	/// (or the history ran out) before every entry was placed; those entries
	615	+	/// are left out, and a later call goes on placing them.
614	616		#[derive(Clone, Debug, Serialize, Deserialize)]
615	617		pub struct LastCommits {
616	618		pub entries: Vec<LastCommit>,