Skip to content
Request access

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

Commit

Merge pull request #256 from flagon-io/255-edge-fail-safe

A broken answer from the git edge never reaches a client: the store answers instead

syntaqxcommitted Parentse892474460fb3fBrowse files
2 files+125−160/2 viewed
+24−13
5252 use std::process::{Command, Stdio};
5353 use std::sync::mpsc::{Receiver, SyncSender, sync_channel};
5454 use std::sync::atomic::{AtomicU64, Ordering};
55−use std::sync::{Arc, Condvar, Mutex, MutexGuard};
55+use std::sync::{Arc, Condvar, Mutex, MutexGuard, RwLock, TryLockError};
5656 use std::time::Instant;
5757
5858 use anyhow::{Context, Result, bail};
117117 root: PathBuf,
118118 state: Mutex<State>,
119119 wake: Condvar,
120− /// Held while the mirror's files change or are read whole: a sync, a
121− /// snapshot being made, one being restored. Never two at once.
122− disk: Mutex<()>,
120+ /// The mirror's files: read (shared) while a git request is answered or
121+ /// a snapshot made from them, written (alone) while a sync or a restore
122+ /// changes them. A sync's fetch and repack replace the packs an
123+ /// upload-pack under way reads, which then answers nothing (git says
124+ /// `expected 'packfile'`): so a sync waits for the answers under way,
125+ /// and a git request that comes while one runs is behind.
126+ disk: RwLock<()>,
123127 }
124128
125129 impl Edge {
315319 }
316320 };
317321 crate::abuse::touch();
318− let disk = edge.disk.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
322+ let disk = edge.disk.write().unwrap_or_else(|poisoned| poisoned.into_inner());
319323 let started = Instant::now();
320324 let result = sync(&edge, &job, || edge.state().changing = true);
321325 drop(disk);
464468 let name = format!("snapshot-{}.tar", SNAPSHOTS.fetch_add(1, Ordering::Relaxed));
465469 let file = edge.root.join(&name);
466470 let made = {
467− let _disk = edge.disk.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
471+ let _disk = edge.disk.read().unwrap_or_else(|poisoned| poisoned.into_inner());
468472 let version = {
469473 let state = edge.state();
470474 if state.changing { None } else { state.version }
523527 respond(request, Response::from_string("x-g1t-edge-version says the snapshot's version").with_status_code(400));
524528 return 400;
525529 };
526− let _disk = edge.disk.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
530+ let _disk = edge.disk.write().unwrap_or_else(|poisoned| poisoned.into_inner());
527531 let mirror = edge.mirror();
528532 {
529533 let state = edge.state();
628632 /// and the body's bytes.
629633 fn serve_git(edge: &Edge, request: Request, endpoint: &str, query: &str) -> (u16, u64) {
630634 crate::abuse::touch();
631− let expected = header(&request, VERSION_HEADER).and_then(|value| value.trim().parse::<u64>().ok());
632− let current = {
633− let state = edge.state();
634− if state.changing { None } else { state.version }
635− };
636635 let behind = |request: Request| {
637636 let behind = Header::from_bytes(EDGE_HEADER.as_bytes(), &b"behind"[..]).expect("a valid header");
638637 respond(request, Response::empty(409).with_header(behind));
639638 (409, 0)
640639 };
640+ // Held until the answer has gone, so no sync replaces the mirror's
641+ // packs under it; while a sync holds the mirror, the request is behind.
642+ let _disk = match edge.disk.try_read() {
643+ Ok(guard) => guard,
644+ Err(TryLockError::Poisoned(poisoned)) => poisoned.into_inner(),
645+ Err(TryLockError::WouldBlock) => return behind(request),
646+ };
647+ let expected = header(&request, VERSION_HEADER).and_then(|value| value.trim().parse::<u64>().ok());
648+ let current = {
649+ let state = edge.state();
650+ if state.changing { None } else { state.version }
651+ };
641652 if expected.is_none() || expected != current {
642653 return behind(request);
643654 }
876887 };
877888 let listening = server.server_addr().to_ip().map_or_else(|| format!("port {port}"), |address| address.to_string());
878889 eprintln!("g1t-runner: gitedge: listening on {listening}, the mirror in {}", root.join(MIRROR).display());
879− let edge = Arc::new(Edge { root, state: Mutex::new(State::default()), wake: Condvar::new(), disk: Mutex::new(()) });
890+ let edge = Arc::new(Edge { root, state: Mutex::new(State::default()), wake: Condvar::new(), disk: RwLock::new(()) });
880891 let worker = edge.clone();
881892 std::thread::spawn(move || work(worker));
882893 for request in server.incoming_requests() {
+101−3
122122 let url = format!("https://runner/gitedge/{repo_id}/{}{query}", git.endpoint);
123123 self.runner.fetch_request(Request::new_with_init(&url, &init)?).await
124124 };
125− match asked.await {
126− Ok(response) if response.status_code() == 200 => Some(response),
127− Ok(_) => None,
125+ let response = match asked.await {
126+ Ok(response) if response.status_code() == 200 => response,
127+ Ok(_) => return None,
128128 Err(error) => {
129129 worker::console_warn!("git edge for {repo_id} not asked: {error}");
130+ return None;
131+ }
132+ };
133+ // A fetch's answer is looked at before it is given: one that is
134+ // empty or starts as no fetch's answer does goes to the store too.
135+ if git.endpoint != "git-upload-pack" || body.is_some_and(lists_refs) {
136+ return Some(response);
137+ }
138+ let length = response.headers().get("content-length").ok().flatten();
139+ match sound(response).await {
140+ Ok(Ok(response)) => Some(response),
141+ Ok(Err(head)) => {
142+ // What it sent, so the log says why (the edge's own log is not kept).
143+ let shown: String = head.iter().take(160).map(|&byte| if byte.is_ascii_graphic() || byte == b' ' { byte as char } else { '.' }).collect();
144+ worker::console_warn!(
145+ "git edge for {repo_id} answered a fetch with nothing it could use: {} bytes before it ended or went wrong (content-length {}): {shown}",
146+ head.len(),
147+ length.as_deref().unwrap_or("none")
148+ );
149+ None
150+ }
151+ Err(error) => {
152+ worker::console_warn!("git edge for {repo_id} answer not read: {error}");
130153 None
131154 }
132155 }
160183 }
161184 }
162185
186+/// Whether an upload-pack request is protocol v2's `ls-refs`, whose answer
187+/// may be a lone flush (a repository with no refs).
188+fn lists_refs(body: &[u8]) -> bool {
189+ body.windows(b"command=ls-refs".len()).any(|window| window == b"command=ls-refs")
190+}
191+
192+/// Whether the start of a fetch's answer is one: its first pkt-line carries
193+/// data (a section, `NAK`, an `ACK`, `shallow`), not an error, and is not a
194+/// flush, a delimiter or the end of the response. `None` until enough of
195+/// it has come to tell.
196+pub(crate) fn starts_soundly(head: &[u8]) -> Option<bool> {
197+ let length = usize::from_str_radix(std::str::from_utf8(head.get(..4)?).ok()?, 16).ok();
198+ let Some(length) = length else { return Some(false) };
199+ if length < 4 {
200+ return Some(false);
201+ }
202+ let payload = &head[4..head.len().min(length)];
203+ if payload.len() < 4.min(length - 4) {
204+ return None;
205+ }
206+ Some(!payload.starts_with(b"ERR "))
207+}
208+
209+/// The edge's answer to a fetch, when it starts as one (`starts_soundly`):
210+/// only its first bytes are read, and the rest streams on behind them.
211+/// What was read of it, when it is empty or starts otherwise. An upload-pack
212+/// that died after `git http-backend` had sent its headers leaves a 200
213+/// with nothing in it, which git takes for `expected 'packfile'`; the
214+/// store answers.
215+async fn sound(mut response: Response) -> worker::Result<std::result::Result<Response, Vec<u8>>> {
216+ let headers = response.headers().clone();
217+ headers.delete("content-length")?;
218+ let mut stream = response.stream()?;
219+ let mut head = Vec::new();
220+ loop {
221+ match starts_soundly(&head) {
222+ Some(true) => break,
223+ Some(false) => return Ok(Err(head)),
224+ None => match futures_util::StreamExt::next(&mut stream).await {
225+ Some(chunk) => head.extend_from_slice(&chunk?),
226+ // All of it has come: it is sound or not as it stands.
227+ None => {
228+ return Ok(match starts_soundly(&head) {
229+ Some(true) => Ok(Response::from_bytes(head)?.with_headers(headers)),
230+ _ => Err(head),
231+ });
232+ }
233+ },
234+ }
235+ }
236+ let rest = futures_util::StreamExt::chain(futures_util::stream::once(async move { Ok::<Vec<u8>, worker::Error>(head) }), stream);
237+ Ok(Ok(Response::from_stream(rest)?.with_headers(headers)))
238+}
239+
163240 #[cfg(test)]
164241 mod tests {
165242 use super::*;
166243
244+ #[test]
245+ fn a_fetchs_answer_starts_with_data() {
246+ // A section, an acknowledgment, NAK, a shallow line: sound.
247+ assert_eq!(starts_soundly(b"000dpackfile\n0027\x02Enumerating"), Some(true));
248+ assert_eq!(starts_soundly(b"0011shallow-info\n"), Some(true));
249+ assert_eq!(starts_soundly(b"0008NAK\n"), Some(true));
250+ assert_eq!(starts_soundly(b"0014acknowledgments\n"), Some(true));
251+ // Nothing, the end of the response, a flush, an error: not.
252+ assert_eq!(starts_soundly(b"0002"), Some(false));
253+ assert_eq!(starts_soundly(b"0000"), Some(false));
254+ assert_eq!(starts_soundly(b"0001"), Some(false));
255+ assert_eq!(starts_soundly(b"0021ERR upload-pack: not our ref\n"), Some(false));
256+ assert_eq!(starts_soundly(b"zzzzpackfile"), Some(false));
257+ // Too little yet to tell.
258+ assert_eq!(starts_soundly(b""), None);
259+ assert_eq!(starts_soundly(b"000d"), None);
260+ assert_eq!(starts_soundly(b"000dpa"), None);
261+ assert!(lists_refs(b"0014command=ls-refs\n0000"));
262+ assert!(!lists_refs(b"0011command=fetch\n0000"));
263+ }
264+
167265 fn git(endpoint: &'static str, service: GitService) -> GitRequest {
168266 GitRequest {
169267 path: g1t_contracts::repos::RepoPath { namespace: "acme".into(), name: "rocket".into() },

Comments

Sign in to comment.