Skip to content
Sign up

flagon-io/g1t

public
0
374 linesCodeBlameRaw
1//! Actions from GitHub, kept on g1t.
2//!
3//! An action `uses:` names that g1t does not host (reach.rs) comes from
4//! GitHub. Before a job fetches it, g1t resolves its ref to a commit, as
5//! GitHub's runner does, from the repository's advertisement
6//! (`info/refs`, which no API rate limit applies to), and keeps that
7//! commit's archive in R2. Every later job that uses the action at that
8//! commit downloads it from g1t: faster than GitHub, the same bytes
9//! whatever happens to the tag, and there while GitHub is not.
10//!
11//! - A ref is looked up again once `REF_FRESH_MS` has passed, so a moved
12//! tag (`v4` moving to a new release) is followed within minutes. When
13//! GitHub does not answer, the commit the ref last named is used, and
14//! the job says so.
15//! - A commit's archive is the one GitHub serves for it, which never
16//! changes. One larger than `ARCHIVE_MAX_BYTES` is not kept: the job
17//! fetches it from GitHub itself.
18//! - An archive no job has used for `ARCHIVE_UNUSED_DAYS` is deleted by
19//! the hourly sweep. Archives are g1t's own, shared by every workspace,
20//! and charged to none.
21
22use futures_util::StreamExt;
23use g1t_contracts::time::rfc3339;
24use g1t_kit::now_ms;
25use serde::Deserialize;
26use serde_json::{Value, json};
27use worker::Result;
28
29use crate::Actions;
30use crate::reach::UsesRef;
31
32/// The largest archive kept: the Worker holds it whole while saving it.
33pub(crate) const ARCHIVE_MAX_BYTES: usize = 32 * 1024 * 1024;
34/// How long a ref's commit is trusted before GitHub is asked again.
35pub(crate) const REF_FRESH_MS: u64 = 5 * 60 * 1000;
36/// An archive unused this long is deleted.
37pub(crate) const ARCHIVE_UNUSED_DAYS: u64 = 30;
38/// An advertisement larger than this is not read: a ref is never worth it.
39const ADVERTISEMENT_MAX_BYTES: usize = 8 * 1024 * 1024;
40const DAY_MS: u64 = 24 * 60 * 60 * 1000;
41
42/// Whether `text` is a whole commit id.
43pub(crate) fn is_sha(text: &str) -> bool {
44 text.len() == 40 && text.chars().all(|c| c.is_ascii_hexdigit())
45}
46
47/// The refs a smart HTTP advertisement (`info/refs?service=git-upload-pack`)
48/// lists, as (commit, name). Peeled tags keep their `^{}`.
49pub(crate) fn parse_advertisement(bytes: &[u8]) -> Vec<(String, String)> {
50 let mut refs = Vec::new();
51 let mut at = 0;
52 while at + 4 <= bytes.len() {
53 let Some(length) = std::str::from_utf8(&bytes[at..at + 4]).ok().and_then(|hex| usize::from_str_radix(hex, 16).ok()) else {
54 break;
55 };
56 // A flush or delimiter packet.
57 if length < 4 {
58 at += 4;
59 continue;
60 }
61 if at + length > bytes.len() {
62 break;
63 }
64 let line = String::from_utf8_lossy(&bytes[at + 4..at + length]);
65 at += length;
66 let line = line.trim_end_matches('\n');
67 if line.starts_with('#') {
68 continue;
69 }
70 // The first ref carries the capabilities after a NUL.
71 let line = line.split('\0').next().unwrap_or_default();
72 if let Some((sha, name)) = line.split_once(' ')
73 && is_sha(sha)
74 {
75 refs.push((sha.to_ascii_lowercase(), name.to_owned()));
76 }
77 }
78 refs
79}
80
81/// The commit `wanted` names among `refs`, as git reads a short name: a
82/// commit id as it is, else a tag, else a branch (`refs/…` names one
83/// exactly). A tag's commit is its peeled one.
84pub(crate) fn pick(refs: &[(String, String)], wanted: &str) -> Option<String> {
85 if is_sha(wanted) {
86 return Some(wanted.to_ascii_lowercase());
87 }
88 let names: Vec<String> = if wanted.starts_with("refs/") {
89 vec![wanted.to_owned()]
90 } else {
91 vec![format!("refs/tags/{wanted}"), format!("refs/heads/{wanted}")]
92 };
93 names.iter().find_map(|name| {
94 let peeled = format!("{name}^{{}}");
95 refs.iter()
96 .find(|(_, listed)| *listed == peeled)
97 .or_else(|| refs.iter().find(|(_, listed)| listed == name))
98 .map(|(sha, _)| sha.clone())
99 })
100}
101
102/// Where a commit's archive is kept in R2.
103pub(crate) fn archive_object(repository: &str, sha: &str) -> String {
104 format!("action-archives/{}/{sha}.tar.gz", repository.to_ascii_lowercase())
105}
106
107/// A GET to GitHub, whose body is read up to `max` bytes. None when it
108/// does not answer 200, or sends more.
109async fn fetch_capped(url: &str, max: usize) -> Option<Vec<u8>> {
110 let headers = worker::Headers::new();
111 // GitHub answers a smart HTTP advertisement to git, which it knows by
112 // its agent.
113 headers.set("user-agent", "git/2.47.0 (g1t)").ok()?;
114 let mut init = worker::RequestInit::new();
115 init.with_method(worker::Method::Get).with_headers(headers);
116 let request = worker::Request::new_with_init(url, &init).ok()?;
117 let mut response = worker::Fetch::Request(request).send().await.ok()?;
118 if response.status_code() != 200 {
119 return None;
120 }
121 let mut stream = response.stream().ok()?;
122 let mut body = Vec::new();
123 while let Some(chunk) = stream.next().await {
124 body.extend_from_slice(&chunk.ok()?);
125 if body.len() > max {
126 return None;
127 }
128 }
129 Some(body)
130}
131
132#[derive(Deserialize)]
133struct RefRow {
134 sha: String,
135 resolved_at: String,
136}
137
138impl Actions {
139 /// Where a job gets `used`, an action g1t does not host:
140 /// `{"source": "github", "sha", "archive", "note"}`. `sha` is the commit
141 /// its ref names (absent when GitHub does not know it, or never
142 /// answered), `archive` whether g1t keeps that commit's archive for the
143 /// job to download, and `note` anything the job should say about it.
144 pub(crate) async fn github_action(&self, used: &UsesRef) -> Result<Value> {
145 let repository = used.full_name().to_ascii_lowercase();
146 let now = now_ms();
147 let (sha, note) = self.resolve_github(&repository, &used.git_ref, now).await?;
148 let Some(sha) = sha else {
149 return Ok(json!({ "source": "github", "note": note }));
150 };
151 let archive = self.keep_archive(&repository, &sha, now).await?;
152 Ok(json!({ "source": "github", "sha": sha, "archive": archive, "note": note }))
153 }
154
155 /// The commit `git_ref` names in GitHub's `repository`, and a note when
156 /// it is the one it last named because GitHub did not answer.
157 async fn resolve_github(&self, repository: &str, git_ref: &str, now: u64) -> Result<(Option<String>, Option<String>)> {
158 if is_sha(git_ref) {
159 return Ok((Some(git_ref.to_ascii_lowercase()), None));
160 }
161 let known = self
162 .db
163 .prepare("SELECT sha, resolved_at FROM action_refs WHERE repository = ? AND ref = ?")
164 .bind(&[repository.into(), git_ref.into()])?
165 .first::<RefRow>(None)
166 .await?;
167 if let Some(row) = &known
168 && row.resolved_at >= rfc3339(now.saturating_sub(REF_FRESH_MS))
169 {
170 return Ok((Some(row.sha.clone()), None));
171 }
172 let url = format!("https://github.com/{repository}.git/info/refs?service=git-upload-pack");
173 match fetch_capped(&url, ADVERTISEMENT_MAX_BYTES).await {
174 Some(advertisement) => {
175 let Some(sha) = pick(&parse_advertisement(&advertisement), git_ref) else {
176 return Ok((None, None));
177 };
178 self.db
179 .prepare(
180 "INSERT INTO action_refs (repository, ref, sha, resolved_at) VALUES (?1, ?2, ?3, ?4)
181 ON CONFLICT (repository, ref) DO UPDATE SET sha = ?3, resolved_at = ?4",
182 )
183 .bind(&[repository.into(), git_ref.into(), sha.as_str().into(), rfc3339(now).into()])?
184 .run()
185 .await?;
186 Ok((Some(sha), None))
187 }
188 None => Ok(match known {
189 Some(row) => {
190 let note = format!(
191 "GitHub did not say which commit {repository}@{git_ref} is; using {}, which it named on {}.",
192 &row.sha,
193 row.resolved_at.get(..10).unwrap_or(&row.resolved_at)
194 );
195 (Some(row.sha), Some(note))
196 }
197 None => (None, None),
198 }),
199 }
200 }
201
202 /// Whether g1t keeps the archive of `repository` at `sha`, fetching it
203 /// from GitHub the first time.
204 async fn keep_archive(&self, repository: &str, sha: &str, now: u64) -> Result<bool> {
205 let Some(bucket) = &self.cache else { return Ok(false) };
206 #[derive(Deserialize)]
207 struct Kept {
208 used_at: String,
209 }
210 let kept = self
211 .db
212 .prepare("SELECT used_at FROM action_archives WHERE repository = ? AND sha = ?")
213 .bind(&[repository.into(), sha.into()])?
214 .first::<Kept>(None)
215 .await?;
216 if let Some(kept) = kept {
217 // Marked used at most once a day: the sweep needs no more.
218 if kept.used_at < rfc3339(now.saturating_sub(DAY_MS)) {
219 self.db
220 .prepare("UPDATE action_archives SET used_at = ? WHERE repository = ? AND sha = ?")
221 .bind(&[rfc3339(now).into(), repository.into(), sha.into()])?
222 .run()
223 .await?;
224 }
225 return Ok(true);
226 }
227 let url = format!("https://codeload.github.com/{repository}/tar.gz/{sha}");
228 let Some(archive) = fetch_capped(&url, ARCHIVE_MAX_BYTES).await else {
229 return Ok(false);
230 };
231 let object = archive_object(repository, sha);
232 let size = archive.len();
233 if let Err(error) = bucket.put(&object, archive).execute().await {
234 worker::console_error!("actions: the archive of {repository}@{sha} was not kept: {error}");
235 return Ok(false);
236 }
237 let at = rfc3339(now);
238 self.db
239 .prepare(
240 "INSERT INTO action_archives (repository, sha, object, size, stored_at, used_at) VALUES (?1, ?2, ?3, ?4, ?5, ?5)
241 ON CONFLICT (repository, sha) DO UPDATE SET used_at = ?5",
242 )
243 .bind(&[repository.into(), sha.into(), object.into(), (size as f64).into(), at.into()])?
244 .run()
245 .await?;
246 Ok(true)
247 }
248
249 /// `job_action_archive`: a running job downloads an action's archive
250 /// g1t keeps, by `report.repository` and `report.sha`. Answers the R2
251 /// object to send, `{"object", "size"}`. Any job may: they are public
252 /// repositories' code.
253 pub async fn job_action_archive(&self, a: g1t_contracts::actions::JobCallArgs) -> Result<g1t_contracts::Outcome<Value>> {
254 crate::check!(self.job_for_token(&a).await?);
255 let repository = a.report["repository"].as_str().unwrap_or_default().to_ascii_lowercase();
256 let sha = a.report["sha"].as_str().unwrap_or_default().to_ascii_lowercase();
257 if !is_sha(&sha) || crate::reach::parse_uses(&format!("{repository}@{sha}")).is_none() {
258 return Ok(crate::fail(g1t_contracts::FailureCode::Invalid, "Name the action's repository as owner/repo, and its commit."));
259 }
260 #[derive(Deserialize)]
261 struct Kept {
262 object: String,
263 size: f64,
264 }
265 let kept = self
266 .db
267 .prepare("SELECT object, size FROM action_archives WHERE repository = ? AND sha = ?")
268 .bind(&[repository.as_str().into(), sha.as_str().into()])?
269 .first::<Kept>(None)
270 .await?;
271 Ok(match kept {
272 Some(kept) => g1t_contracts::Outcome::Ok(json!({ "object": kept.object, "size": kept.size })),
273 None => crate::fail(g1t_contracts::FailureCode::NotFound, format!("g1t does not keep {repository}@{sha}.")),
274 })
275 }
276
277 /// Hourly: archives no job has used for `ARCHIVE_UNUSED_DAYS`, and refs
278 /// not looked up as long.
279 pub async fn sweep_action_archives(&self, now: u64) -> Result<()> {
280 let before = rfc3339(now.saturating_sub(ARCHIVE_UNUSED_DAYS * DAY_MS));
281 #[derive(Deserialize)]
282 struct Gone {
283 repository: String,
284 sha: String,
285 object: String,
286 }
287 let gone = self
288 .db
289 .prepare("SELECT repository, sha, object FROM action_archives WHERE used_at < ? LIMIT 100")
290 .bind(&[before.as_str().into()])?
291 .all()
292 .await?
293 .results::<Gone>()?;
294 if let Some(bucket) = &self.cache
295 && !gone.is_empty()
296 {
297 bucket.delete_multiple(gone.iter().map(|g| g.object.as_str()).collect()).await?;
298 for archive in &gone {
299 self.db
300 .prepare("DELETE FROM action_archives WHERE repository = ? AND sha = ?")
301 .bind(&[archive.repository.as_str().into(), archive.sha.as_str().into()])?
302 .run()
303 .await?;
304 }
305 }
306 self.db.prepare("DELETE FROM action_refs WHERE resolved_at < ?").bind(&[before.into()])?.run().await?;
307 Ok(())
308 }
309}
310
311#[cfg(test)]
312mod tests {
313 use super::*;
314
315 fn pkt(payload: &str) -> Vec<u8> {
316 let mut out = format!("{:04x}", payload.len() + 4).into_bytes();
317 out.extend_from_slice(payload.as_bytes());
318 out
319 }
320
321 const MAIN: &str = "1111111111111111111111111111111111111111";
322 const V4_TAG: &str = "2222222222222222222222222222222222222222";
323 const V4_COMMIT: &str = "3333333333333333333333333333333333333333";
324 const V1_BRANCH: &str = "4444444444444444444444444444444444444444";
325 const V1_TAG: &str = "5555555555555555555555555555555555555555";
326
327 fn advertisement() -> Vec<u8> {
328 let mut bytes = pkt("# service=git-upload-pack\n");
329 bytes.extend_from_slice(b"0000");
330 bytes.extend(pkt(&format!("{MAIN} HEAD\0multi_ack thin-pack symref=HEAD:refs/heads/main\n")));
331 bytes.extend(pkt(&format!("{MAIN} refs/heads/main\n")));
332 bytes.extend(pkt(&format!("{V1_BRANCH} refs/heads/v1\n")));
333 bytes.extend(pkt(&format!("{V1_TAG} refs/tags/v1\n")));
334 bytes.extend(pkt(&format!("{V4_TAG} refs/tags/v4\n")));
335 bytes.extend(pkt(&format!("{V4_COMMIT} refs/tags/v4^{{}}\n")));
336 bytes.extend_from_slice(b"0000");
337 bytes
338 }
339
340 #[test]
341 fn an_advertisement_lists_its_refs_capabilities_aside() {
342 let refs = parse_advertisement(&advertisement());
343 assert_eq!(refs.len(), 6);
344 assert_eq!(refs[0], (MAIN.to_owned(), "HEAD".to_owned()));
345 assert_eq!(refs[5], (V4_COMMIT.to_owned(), "refs/tags/v4^{}".to_owned()));
346 // Cut short: what came whole is kept.
347 let whole = advertisement();
348 assert_eq!(parse_advertisement(&whole[..whole.len() - 30]).len(), 5);
349 assert!(parse_advertisement(b"not git").is_empty());
350 }
351
352 #[test]
353 fn a_ref_is_read_as_git_reads_a_short_name() {
354 let refs = parse_advertisement(&advertisement());
355 // A tag's commit, not the tag object.
356 assert_eq!(pick(&refs, "v4").as_deref(), Some(V4_COMMIT));
357 // A tag before a branch of the same name.
358 assert_eq!(pick(&refs, "v1").as_deref(), Some(V1_TAG));
359 assert_eq!(pick(&refs, "refs/heads/v1").as_deref(), Some(V1_BRANCH));
360 assert_eq!(pick(&refs, "main").as_deref(), Some(MAIN));
361 // A commit is itself, known or not.
362 let other = "ABCDEF0123456789ABCDEF0123456789ABCDEF01";
363 assert_eq!(pick(&refs, other).as_deref(), Some("abcdef0123456789abcdef0123456789abcdef01"));
364 assert_eq!(pick(&refs, "v9"), None);
365 assert_eq!(pick(&refs, "abc123"), None);
366 }
367
368 #[test]
369 fn archives_are_kept_by_repository_and_commit() {
370 assert_eq!(archive_object("Actions/Checkout", V4_COMMIT), format!("action-archives/actions/checkout/{V4_COMMIT}.tar.gz"));
371 assert!(is_sha(MAIN));
372 assert!(!is_sha("v4"));
373 }
374}