Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp
430 linesCodeBlameRaw
1//! Pull requests' working copies after their pull request is done.
2//!
3//! Every pull request made from an issue works in its own copy of the
4//! repository in the git store (`pulls--<pull id>`, `fork_for_pull`).
5//! Cloudflare does not say whether a fork shares objects with its source,
6//! and storage is billed and capped per account (1 TB), so copies are not
7//! kept forever: [`retention_days`] after the pull request merges or
8//! closes (`pull.merged`, `pull.closed`), the hourly sweep removes the
9//! copy's git data. Reopened within the window (`pull.reopened`), it stays.
10//!
11//! Nothing is lost. Before the copy goes, its head is kept in the
12//! repository it came from as `refs/pull/<pull id>/head` (only the
13//! objects the repository lacks travel), and the row stays, with that
14//! commit. Reads of the copy (the pull request's changes, its files) are
15//! answered from the repository ([`Viewed`]). Anything that writes to it
16//! (a push, git access, catching up, landing) makes it again first, from
17//! that ref ([`Repos::live`]).
18
19use g1t_contracts::repos::{Branch, Commit, GitAccess, Repo, TreeEntry};
20use g1t_contracts::time::rfc3339;
21use g1t_kit::now_ms;
22use worker::wasm_bindgen::JsValue;
23use worker::{Env, Result};
24
25use crate::registry::{Registry, RepoRow, store_key};
26use crate::store::{GitRepo, GitStore, Scope};
27use crate::{PULLS_NAMESPACE, Repos, land, refs};
28
29/// Days a working copy is kept after its pull request merges or closes,
30/// unless `FORK_RETENTION_DAYS` says otherwise.
31pub const DEFAULT_RETENTION_DAYS: u64 = 7;
32/// How many working copies one sweep removes.
33const RETIRES_PER_SWEEP: u32 = 25;
34
35pub fn retention_days(env: &Env) -> u64 {
36 env.var("FORK_RETENTION_DAYS")
37 .ok()
38 .and_then(|value| value.to_string().parse().ok())
39 .unwrap_or(DEFAULT_RETENTION_DAYS)
40}
41
42/// When a working copy whose pull request settled at `now` is removed.
43pub fn retire_after(now: u64, days: u64) -> String {
44 rfc3339(now + days * 24 * 3600 * 1000)
45}
46
47/// Where a pull request's head is kept in its repository.
48pub fn pull_ref(pull_id: &str) -> String {
49 format!("refs/pull/{pull_id}/head")
50}
51
52/// What a pull request event means for its working copy.
53#[derive(Debug, PartialEq, Eq)]
54pub enum PullChange {
55 /// Merged or closed: remove it after the window.
56 Settled,
57 /// Open again: keep it, or make it again.
58 Reopened,
59}
60
61pub fn pull_change(kind: &str) -> Option<PullChange> {
62 match kind {
63 "pull.merged" | "pull.closed" => Some(PullChange::Settled),
64 "pull.reopened" => Some(PullChange::Reopened),
65 _ => None,
66 }
67}
68
69/// The pull request an event is about.
70pub fn pull_id_of(data: &serde_json::Value) -> Option<String> {
71 data.get("pullId").or_else(|| data.get("pull_id")).and_then(|id| id.as_str()).map(str::to_owned)
72}
73
74/// A repository as reads see it: a working copy that was removed reads
75/// as the repository it came from, with its branch at the head it had.
76pub struct Viewed<R> {
77 inner: R,
78 /// The copy's branch, and the commit it pointed to.
79 alias: Option<(String, String)>,
80}
81
82impl<R> Viewed<R> {
83 pub fn plain(inner: R) -> Self {
84 Viewed { inner, alias: None }
85 }
86
87 pub fn retired(inner: R, branch: &str, head: &str) -> Self {
88 Viewed { inner, alias: Some((branch.to_owned(), head.to_owned())) }
89 }
90
91 /// `git_ref`, or the head a removed copy's branch pointed to.
92 fn resolve<'a>(&'a self, git_ref: &'a str) -> &'a str {
93 match &self.alias {
94 Some((branch, head)) if aliases(git_ref, branch) => head,
95 _ => git_ref,
96 }
97 }
98}
99
100/// Whether `git_ref` names `branch`, or HEAD.
101fn aliases(git_ref: &str, branch: &str) -> bool {
102 git_ref == branch || git_ref == "HEAD" || git_ref.strip_prefix("refs/heads/") == Some(branch)
103}
104
105impl<R: GitRepo> GitRepo for Viewed<R> {
106 async fn access(&self, scope: Scope) -> Result<GitAccess> {
107 self.inner.access(scope).await
108 }
109
110 async fn branches(&self) -> Result<Vec<Branch>> {
111 match &self.alias {
112 Some((branch, head)) => Ok(vec![Branch { name: branch.clone(), hash: head.clone() }]),
113 None => self.inner.branches().await,
114 }
115 }
116
117 async fn log(&self, git_ref: &str, limit: u32) -> Result<Vec<Commit>> {
118 self.inner.log(self.resolve(git_ref), limit).await
119 }
120
121 async fn parents(&self, commit_hash: &str) -> Result<Option<Vec<String>>> {
122 self.inner.parents(commit_hash).await
123 }
124
125 async fn read_tree(&self, tree_hash: &str) -> Result<Option<Vec<TreeEntry>>> {
126 self.inner.read_tree(tree_hash).await
127 }
128
129 async fn read_blob(&self, blob_hash: &str) -> Result<Option<Vec<u8>>> {
130 self.inner.read_blob(blob_hash).await
131 }
132
133 /// The store's own, which keeps sizes by hash without reading bytes.
134 async fn blob_size(&self, blob_hash: &str) -> Result<Option<u64>> {
135 self.inner.blob_size(blob_hash).await
136 }
137
138 async fn read_file(&self, git_ref: &str, path: &str) -> Result<Option<Vec<u8>>> {
139 self.inner.read_file(self.resolve(git_ref), path).await
140 }
141
142 async fn fork(&self, target_key: &str, default_branch_only: bool) -> Result<()> {
143 if self.alias.is_some() {
144 return Err(worker::Error::RustError("a removed working copy is not forked".into()));
145 }
146 self.inner.fork(target_key, default_branch_only).await
147 }
148
149 fn at_refs_version(&mut self, version: Option<u64>) {
150 self.inner.at_refs_version(version);
151 }
152}
153
154impl Registry {
155 /// The working copy of a pull request, if it has one.
156 pub async fn pull_fork(&self, pull_id: &str) -> Result<Option<Repo>> {
157 Ok(self
158 .db
159 .prepare("SELECT * FROM repos WHERE namespace = ? AND name = ? AND fork_of IS NOT NULL AND deleted_at IS NULL")
160 .bind(&[PULLS_NAMESPACE.into(), pull_id.to_lowercase().into()])?
161 .first::<RepoRow>(None)
162 .await?
163 .map(Repo::from))
164 }
165
166 /// Sets, or with `None` clears, when a working copy is removed.
167 pub async fn set_retire_after(&self, id: &str, after: Option<&str>) -> Result<()> {
168 self.db
169 .prepare("UPDATE repos SET retire_after = ? WHERE id = ? AND retired_at IS NULL")
170 .bind(&[after.map_or(JsValue::NULL, JsValue::from), id.into()])?
171 .run()
172 .await?;
173 Ok(())
174 }
175
176 /// Working copies whose time has come, oldest first.
177 pub async fn retiring(&self, now: &str, limit: u32) -> Result<Vec<Repo>> {
178 Ok(self
179 .db
180 .prepare(
181 "SELECT * FROM repos
182 WHERE retire_after IS NOT NULL AND retire_after <= ? AND retired_at IS NULL
183 AND fork_of IS NOT NULL AND deleted_at IS NULL
184 ORDER BY retire_after LIMIT ?",
185 )
186 .bind(&[now.into(), limit.into()])?
187 .all()
188 .await?
189 .results::<RepoRow>()?
190 .into_iter()
191 .map(Repo::from)
192 .collect())
193 }
194
195 /// Records that a working copy's git data is gone, and the head it had;
196 /// `None` puts it back as it was, for a removal that failed.
197 pub async fn set_retired(&self, id: &str, retired: Option<(&str, Option<&str>)>) -> Result<()> {
198 let (at, head) = match retired {
199 Some((at, head)) => (JsValue::from(at), head.map_or(JsValue::NULL, JsValue::from)),
200 None => (JsValue::NULL, JsValue::NULL),
201 };
202 self.db
203 .prepare("UPDATE repos SET retired_at = ?, retired_head = ? WHERE id = ?")
204 .bind(&[at, head, id.into()])?
205 .run()
206 .await?;
207 crate::registry::note_retired(id, retired.and_then(|(_, head)| head));
208 Ok(())
209 }
210
211 /// Records that a working copy was made again; `retire_after`, when to
212 /// remove it next.
213 pub async fn revived(&self, id: &str, retire_after: Option<&str>) -> Result<()> {
214 self.db
215 .prepare("UPDATE repos SET retired_at = NULL, retired_head = NULL, retire_after = ? WHERE id = ?")
216 .bind(&[retire_after.map_or(JsValue::NULL, JsValue::from), id.into()])?
217 .run()
218 .await?;
219 crate::registry::note_retired(id, None);
220 Ok(())
221 }
222}
223
224impl<S: GitStore> Repos<S> {
225 /// The repository's store as reads see it: a removed working copy
226 /// reads from the repository it came from. Reads by branch are kept
227 /// under its refs version (store.rs).
228 pub(crate) async fn read_git(&self, repo: &Repo) -> Result<Viewed<S::Repo>> {
229 let now = now_ms();
230 if let (Some(head), Some(parent_id)) = (crate::registry::retired(&repo.id), &repo.fork_of)
231 && let Some(parent) = self.registry.by_id(parent_id).await?
232 {
233 let mut git = self.store.open(&store_key(&parent)).await?;
234 git.at_refs_version(crate::refs_cache::usable(crate::registry::refs_state(&parent.id), now));
235 return Ok(Viewed::retired(git, &repo.default_branch, &head));
236 }
237 let mut git = self.store.open(&store_key(repo)).await?;
238 git.at_refs_version(crate::refs_cache::usable(crate::registry::refs_state(&repo.id), now));
239 Ok(Viewed::plain(git))
240 }
241
242 /// Makes a removed working copy again before anything writes to it.
243 /// Its pull request is still settled, so it is removed again after the
244 /// window unless the pull request reopens.
245 pub(crate) async fn live(&self, repo: &Repo) -> Result<()> {
246 if crate::registry::retired(&repo.id).is_none() {
247 return Ok(());
248 }
249 self.revive(repo, Some(retire_after(now_ms(), self.fork_days))).await
250 }
251
252 /// `pull.merged` or `pull.closed`: the working copy goes in `days`.
253 pub(crate) async fn pull_settled(&self, pull_id: &str) -> Result<()> {
254 if let Some(fork) = self.registry.pull_fork(pull_id).await? {
255 self.registry.set_retire_after(&fork.id, Some(&retire_after(now_ms(), self.fork_days))).await?;
256 }
257 Ok(())
258 }
259
260 /// `pull.reopened`: the working copy stays, made again if it had gone.
261 pub(crate) async fn pull_reopened(&self, pull_id: &str) -> Result<()> {
262 let Some(fork) = self.registry.pull_fork(pull_id).await? else {
263 return Ok(());
264 };
265 if crate::registry::retired(&fork.id).is_some() {
266 return self.revive(&fork, None).await;
267 }
268 self.registry.set_retire_after(&fork.id, None).await
269 }
270
271 /// The sweep: removes the working copies whose time has come.
272 pub(crate) async fn retire_due(&self) -> Result<u32> {
273 let due = self.registry.retiring(&rfc3339(now_ms()), RETIRES_PER_SWEEP).await?;
274 let mut retired = 0;
275 for fork in due {
276 match self.retire(&fork).await {
277 Ok(()) => retired += 1,
278 Err(error) => worker::console_error!("working copy {} not removed: {error}", fork.id),
279 }
280 }
281 Ok(retired)
282 }
283
284 /// Keeps a working copy's head in its repository, then removes its git
285 /// data. A failure before the removal leaves everything as it was.
286 async fn retire(&self, fork: &Repo) -> Result<()> {
287 // Being copied to another namespace (moves.rs): the next sweep.
288 if let Some(reason) = crate::registry::paused(&fork.id, now_ms()) {
289 return Err(worker::Error::RustError(format!("paused: {reason}")));
290 }
291 let key = store_key(fork);
292 let parent = match &fork.fork_of {
293 Some(id) => self.registry.by_id(id).await?,
294 None => None,
295 };
296 // Already gone from the store: it says so when first asked.
297 let read = match self.store.open(&key).await {
298 Ok(git) => git.log(&fork.default_branch, 1).await,
299 Err(error) => Err(error),
300 };
301 let head = match read {
302 Ok(commits) => commits.into_iter().next().map(|commit| commit.hash),
303 Err(error) if error.to_string().contains("NOT_FOUND") => None,
304 Err(error) => return Err(error),
305 };
306 if let (Some(parent), Some(head)) = (&parent, &head) {
307 self.keep_head(fork, parent, head).await?;
308 }
309 let now = rfc3339(now_ms());
310 self.registry.set_retired(&fork.id, Some((&now, head.as_deref()))).await?;
311 if let Err(error) = self.store.delete(&key).await {
312 // Back as it was, for the next sweep.
313 self.registry.set_retired(&fork.id, None).await?;
314 return Err(error);
315 }
316 Ok(())
317 }
318
319 /// Points `refs/pull/<pull id>/head` in `parent` at `head`, sending
320 /// only the objects the parent lacks.
321 async fn keep_head(&self, fork: &Repo, parent: &Repo, head: &str) -> Result<()> {
322 let parent_git = self.store.open(&store_key(parent)).await?;
323 let reference = pull_ref(&fork.name);
324 let parent_read = parent_git.access(Scope::Read).await?;
325 let refs = refs::all(&parent_read).await?;
326 let existing = refs.iter().find(|(name, _)| *name == reference).map(|(_, hash)| hash.clone());
327 if existing.as_deref() == Some(head) {
328 return Ok(());
329 }
330 let has_it = !parent_git.log(head, 1).await?.is_empty();
331 let pack = if has_it {
332 land::EMPTY_PACK.to_vec()
333 } else {
334 let base = refs.iter().find(|(name, _)| *name == format!("refs/heads/{}", parent.default_branch)).map(|(_, hash)| hash.clone());
335 let fork_read = self.store.open(&store_key(fork)).await?.access(Scope::Read).await?;
336 land::fetch_pack(&fork_read, head, base.as_deref()).await?
337 };
338 let parent_write = parent_git.access(Scope::Write).await?;
339 let kept = land::push_ref(&parent_write, &reference, existing.as_deref(), head, Some(pack)).await;
340 self.refs_moved(&parent.id).await;
341 kept?.map_err(|reason| worker::Error::RustError(format!("{reference} not kept in {}: {reason}", parent.id)))
342 }
343
344 /// Makes a removed working copy again: a fork of its repository, with
345 /// its branch moved back to the head it had.
346 async fn revive(&self, fork: &Repo, retire_after: Option<String>) -> Result<()> {
347 let head = crate::registry::retired(&fork.id);
348 let parent = match &fork.fork_of {
349 Some(id) => self.registry.by_id(id).await?,
350 None => None,
351 };
352 let (Some(head), Some(parent)) = (head, parent) else {
353 return Err(worker::Error::RustError(format!(
354 "working copy {} cannot be made again: its repository or head is gone",
355 fork.id
356 )));
357 };
358 let key = store_key(fork);
359 let parent_git = self.store.open(&store_key(&parent)).await?;
360 parent_git.fork(&key, true).await?;
361 let git = self.store.open(&key).await?;
362 let current = git.log(&fork.default_branch, 1).await?.into_iter().next().map(|commit| commit.hash);
363 if current.as_deref() != Some(head.as_str()) {
364 let pack = land::fetch_pack(&parent_git.access(Scope::Read).await?, &head, current.as_deref()).await?;
365 let moved = land::push_ref(
366 &git.access(Scope::Write).await?,
367 &format!("refs/heads/{}", fork.default_branch),
368 current.as_deref(),
369 &head,
370 Some(pack),
371 )
372 .await;
373 self.refs_moved(&fork.id).await;
374 moved?.map_err(|reason| worker::Error::RustError(format!("working copy {} made again, but not at {head}: {reason}", fork.id)))?;
375 }
376 self.registry.revived(&fork.id, retire_after.as_deref()).await?;
377 self.refs_moved(&fork.id).await;
378 Ok(())
379 }
380}
381
382#[cfg(test)]
383mod tests {
384 use super::*;
385
386 #[test]
387 fn every_event_that_settles_a_copy_is_sent_to_this_queue() {
388 let users = ["user.deleting", "user.restored"];
389 for kind in g1t_contracts::webhooks::EVENT_TYPES.iter().chain(users.iter()) {
390 let acts = pull_change(kind).is_some()
391 || crate::stats::shown_differently(kind, &serde_json::json!({ "username": "ana" })).is_some();
392 if acts {
393 assert!(g1t_contracts::subscribers::routed("SUBSCRIBER_REPOS", kind), "{kind} is not routed to repos");
394 }
395 }
396 for kind in ["workspace.renamed", "workspace.deleting", "workspace.restored", "workspace.deleted"] {
397 assert!(g1t_contracts::subscribers::routed("SUBSCRIBER_REPOS", kind));
398 }
399 }
400
401 #[test]
402 fn a_copy_is_removed_days_after_its_pull_request_settles() {
403 let settled = 1_791_936_000_000; // 2026-10-14T00:00:00Z
404 assert_eq!(retire_after(settled, 7), "2026-10-21T00:00:00.000Z");
405 assert_eq!(retire_after(settled, 0), "2026-10-14T00:00:00.000Z");
406 // The sweep compares these as text, which orders like time.
407 assert!(retire_after(settled, 7) > retire_after(settled, 6));
408 assert_eq!(pull_change("pull.merged"), Some(PullChange::Settled));
409 assert_eq!(pull_change("pull.closed"), Some(PullChange::Settled));
410 assert_eq!(pull_change("pull.reopened"), Some(PullChange::Reopened));
411 assert_eq!(pull_change("pull.updated"), None);
412 assert_eq!(pull_id_of(&serde_json::json!({ "pullId": "pul_1", "number": 3 })).as_deref(), Some("pul_1"));
413 assert_eq!(pull_id_of(&serde_json::json!({ "pull_id": "pul_2" })).as_deref(), Some("pul_2"));
414 assert_eq!(pull_id_of(&serde_json::json!({})), None);
415 assert_eq!(pull_ref("pul_1"), "refs/pull/pul_1/head");
416 }
417
418 #[test]
419 fn a_removed_copy_reads_its_branch_at_the_head_it_had() {
420 let viewed = Viewed::retired((), "main", &"a".repeat(40));
421 assert_eq!(viewed.resolve("main"), "a".repeat(40));
422 assert_eq!(viewed.resolve("refs/heads/main"), "a".repeat(40));
423 assert_eq!(viewed.resolve("HEAD"), "a".repeat(40));
424 // Anything else is read from the repository as it is.
425 assert_eq!(viewed.resolve("dev"), "dev");
426 assert_eq!(viewed.resolve(&"b".repeat(40)), "b".repeat(40));
427 let plain = Viewed::plain(());
428 assert_eq!(plain.resolve("main"), "main");
429 }
430}