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
270 linesCodeBlameRaw
1//! `pushed`: what a push to a branch brought into it. Work asks after each
2//! push, to mark merged the pull requests whose heads it brought in, as a
3//! merge from the site would, and to close the issues its commits' messages
4//! name after a closing keyword (work's pushed.rs).
5//!
6//! One walk from the new head, every parent followed, stopping at what the
7//! branch held before: its first-parent line and those commits' parents,
8//! read in one `log`. A commit off that line that the branch already held
9//! (a side branch merged long ago) can be taken for new when the push
10//! merges it again; the pull request heads found are therefore checked
11//! against the old head exactly before they are said to have landed.
12
13use std::collections::{HashMap, HashSet, VecDeque};
14
15use g1t_contracts::repos::{Commit, MAX_PUSHED, Pushed, PushedArgs, PushedCommit};
16use worker::Result;
17
18use crate::store::{GitRepo, GitStore};
19use crate::{MAX_ANCESTRY, Repos, descends_from};
20
21/// How many commits of a side line one read brings: a merged branch's
22/// commits come in one `log` rather than one read each.
23const SIDE_READ: u32 = 50;
24
25impl<S: GitStore> Repos<S> {
26 pub(crate) async fn pushed(&self, a: PushedArgs) -> Result<Pushed> {
27 let Some(repo) = self.registry.by_id(&a.repo_id).await? else {
28 return Ok(Pushed::default());
29 };
30 let git = self.read_git(&repo).await?;
31 brought_in(&git, a).await
32 }
33}
34
35/// [`Repos::pushed`] on one repository's commits.
36pub(crate) async fn brought_in<R: GitRepo>(git: &R, a: PushedArgs) -> Result<Pushed> {
37 let before = match &a.before {
38 Some(before) => git.log(before, MAX_ANCESTRY).await?,
39 None => Vec::new(),
40 };
41 // What the branch held: its line, and where each commit on it came from.
42 let mut held: HashSet<&str> = HashSet::new();
43 for commit in &before {
44 held.insert(commit.hash.as_str());
45 held.extend(commit.parents.iter().map(String::as_str));
46 }
47 let mut known: HashMap<String, Commit> = git
48 .log(&a.after, MAX_PUSHED as u32 + 1)
49 .await?
50 .into_iter()
51 .map(|commit| (commit.hash.clone(), commit))
52 .collect();
53 let mut seen: HashSet<String> = HashSet::new();
54 let mut new: Vec<Commit> = Vec::new();
55 let mut queue: VecDeque<String> = VecDeque::from([a.after.clone()]);
56 while let Some(hash) = queue.pop_front() {
57 if held.contains(hash.as_str()) || !seen.insert(hash.clone()) {
58 continue;
59 }
60 if new.len() >= MAX_ANCESTRY as usize {
61 break;
62 }
63 if !known.contains_key(&hash) {
64 for commit in git.log(&hash, SIDE_READ).await? {
65 known.entry(commit.hash.clone()).or_insert(commit);
66 }
67 }
68 let Some(commit) = known.get(&hash) else { continue };
69 queue.extend(commit.parents.iter().cloned());
70 new.push(commit.clone());
71 }
72 let brought: HashSet<&str> = new.iter().map(|commit| commit.hash.as_str()).collect();
73 let mut landed = Vec::new();
74 for head in &a.heads {
75 if landed.contains(head) || !brought.contains(head.as_str()) {
76 continue;
77 }
78 // Taken for new, but held all along off the old line.
79 if !before.is_empty() && descends_from(git, &before, head).await? {
80 continue;
81 }
82 landed.push(head.clone());
83 }
84 // A new branch brought nothing to tell anyone about: its commits were
85 // pushed elsewhere first, or are the repository's first.
86 let commits = if a.messages && a.before.is_some() {
87 new.iter()
88 .take(MAX_PUSHED)
89 .rev()
90 .map(|commit| PushedCommit { hash: commit.hash.clone(), message: commit.message.clone() })
91 .collect()
92 } else {
93 Vec::new()
94 };
95 Ok(Pushed { landed, commits })
96}
97
98#[cfg(test)]
99mod tests {
100 use std::cell::RefCell;
101 use std::future::Future;
102 use std::pin::pin;
103 use std::task::{Context, Poll, Waker};
104
105 use g1t_contracts::repos::{Branch, GitAccess, Signature, TreeEntry};
106
107 use super::*;
108 use crate::store::Scope;
109
110 fn run<F: Future>(future: F) -> F::Output {
111 match pin!(future).as_mut().poll(&mut Context::from_waker(Waker::noop())) {
112 Poll::Ready(output) => output,
113 Poll::Pending => panic!("the fake store never waits"),
114 }
115 }
116
117 fn commit(hash: &str, parents: &[&str]) -> Commit {
118 Commit {
119 hash: hash.into(),
120 tree_hash: format!("t{hash}"),
121 message: format!("commit {hash}"),
122 author: Signature { name: "a".into(), email: "a@example.com".into() },
123 parents: parents.iter().map(|&p| p.to_owned()).collect(),
124 authored_at: "2026-10-14T00:00:00.000Z".into(),
125 }
126 }
127
128 /// Commits by hash; `log` follows first parents only, as the store
129 /// does, and `parents` answers for any commit. Counts `log` reads.
130 #[derive(Default)]
131 struct Fake {
132 commits: HashMap<String, Commit>,
133 logs: RefCell<u32>,
134 }
135
136 impl Fake {
137 fn with(commits: Vec<Commit>) -> Fake {
138 Fake { commits: commits.into_iter().map(|c| (c.hash.clone(), c)).collect(), ..Fake::default() }
139 }
140 }
141
142 impl GitRepo for Fake {
143 async fn access(&self, _scope: Scope) -> Result<GitAccess> {
144 unimplemented!()
145 }
146 async fn branches(&self) -> Result<Vec<Branch>> {
147 Ok(Vec::new())
148 }
149 async fn log(&self, git_ref: &str, limit: u32) -> Result<Vec<Commit>> {
150 *self.logs.borrow_mut() += 1;
151 let mut out = Vec::new();
152 let mut at = self.commits.get(git_ref);
153 while let Some(commit) = at {
154 if out.len() as u32 >= limit {
155 break;
156 }
157 out.push(commit.clone());
158 at = commit.parents.first().and_then(|parent| self.commits.get(parent));
159 }
160 Ok(out)
161 }
162 async fn parents(&self, commit_hash: &str) -> Result<Option<Vec<String>>> {
163 Ok(self.commits.get(commit_hash).map(|commit| commit.parents.clone()))
164 }
165 async fn read_tree(&self, _tree_hash: &str) -> Result<Option<Vec<TreeEntry>>> {
166 Ok(None)
167 }
168 async fn read_blob(&self, _blob_hash: &str) -> Result<Option<Vec<u8>>> {
169 Ok(None)
170 }
171 async fn read_file(&self, _git_ref: &str, _path: &str) -> Result<Option<Vec<u8>>> {
172 Ok(None)
173 }
174 async fn fork(&self, _target_key: &str, _default_branch_only: bool) -> Result<()> {
175 Ok(())
176 }
177 }
178
179 fn args(before: Option<&str>, after: &str, heads: &[&str]) -> PushedArgs {
180 PushedArgs {
181 repo_id: "rep_1".into(),
182 before: before.map(str::to_owned),
183 after: after.into(),
184 heads: heads.iter().map(|&h| h.to_owned()).collect(),
185 messages: true,
186 }
187 }
188
189 fn hashes(pushed: &Pushed) -> Vec<&str> {
190 pushed.commits.iter().map(|c| c.hash.as_str()).collect()
191 }
192
193 /// m1 ← m2 on main; f1 ← f2 a branch off m2.
194 fn history() -> Vec<Commit> {
195 vec![commit("m1", &[]), commit("m2", &["m1"]), commit("f1", &["m2"]), commit("f2", &["f1"])]
196 }
197
198 #[test]
199 fn a_fast_forward_brings_in_the_branch_and_its_head() {
200 let git = Fake::with(history());
201 let pushed = run(brought_in(&git, args(Some("m2"), "f2", &["f2", "f1", "m2"]))).unwrap();
202 // f1 is a head too (a pull request stacked under the other).
203 assert_eq!(pushed.landed, vec!["f2", "f1"]);
204 assert_eq!(hashes(&pushed), vec!["f1", "f2"]);
205 }
206
207 #[test]
208 fn a_merge_commit_brings_in_its_second_parent() {
209 let mut commits = history();
210 commits.extend([commit("m3", &["m2"]), commit("x", &["m3", "f2"])]);
211 let git = Fake::with(commits);
212 let pushed = run(brought_in(&git, args(Some("m3"), "x", &["f2"]))).unwrap();
213 assert_eq!(pushed.landed, vec!["f2"]);
214 assert_eq!(hashes(&pushed), vec!["f1", "f2", "x"]);
215 }
216
217 #[test]
218 fn a_head_the_branch_already_held_did_not_land() {
219 // g ← h merged into main long ago (m3); main went on to m4. A pull
220 // request whose head is still g, and a push that merges g again.
221 let commits = vec![
222 commit("m1", &[]),
223 commit("g", &["m1"]),
224 commit("h", &["g"]),
225 commit("m2", &["m1"]),
226 commit("m3", &["m2", "h"]),
227 commit("m4", &["m3"]),
228 commit("y", &["m4", "g"]),
229 ];
230 let git = Fake::with(commits);
231 let pushed = run(brought_in(&git, args(Some("m4"), "y", &["g", "m4"]))).unwrap();
232 assert!(pushed.landed.is_empty());
233 // A head that was never pushed elsewhere is not looked for at all.
234 assert!(!hashes(&pushed).contains(&"m4"));
235 }
236
237 #[test]
238 fn nothing_new_lands_nothing() {
239 let git = Fake::with(history());
240 let pushed = run(brought_in(&git, args(Some("f2"), "f2", &["f2"]))).unwrap();
241 assert!(pushed.landed.is_empty() && pushed.commits.is_empty());
242 // Moving back (a force push to an older commit) brings nothing in.
243 let back = run(brought_in(&git, args(Some("f2"), "m2", &["m2"]))).unwrap();
244 assert!(back.landed.is_empty() && back.commits.is_empty());
245 }
246
247 #[test]
248 fn a_new_branch_lists_no_messages() {
249 let git = Fake::with(history());
250 let pushed = run(brought_in(&git, args(None, "f2", &["f2"]))).unwrap();
251 assert_eq!(pushed.landed, vec!["f2"]);
252 assert!(pushed.commits.is_empty());
253 }
254
255 #[test]
256 fn a_merged_branch_is_read_in_one_go() {
257 // Twenty commits on a side branch merged in: one read of main and
258 // one of the side line, not one per commit.
259 let mut commits = vec![commit("m1", &[]), commit("m2", &["m1"])];
260 commits.push(commit("s1", &["m1"]));
261 for i in 2..=20 {
262 commits.push(commit(&format!("s{i}"), &[&format!("s{}", i - 1)]));
263 }
264 commits.push(commit("x", &["m2", "s20"]));
265 let git = Fake::with(commits);
266 let pushed = run(brought_in(&git, args(Some("m2"), "x", &[]))).unwrap();
267 assert_eq!(pushed.commits.len(), 21);
268 assert_eq!(*git.logs.borrow(), 3);
269 }
270}