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

Commit

Merge pull request #177 from flagon-io/pack-memory-budget

Read a pushed pack in place, so a large push is scanned instead of answering 500

syntaqxcommitted Parentsfc98f0773ab873Browse files
4 files+265−750/4 viewed
+256−66
99 //! of whole objects ([`write_pack`]), or one fetched from elsewhere with a
1010 //! few objects added ([`extend_pack`]).
1111
12−use std::collections::HashMap;
12+use std::cell::RefCell;
13+use std::collections::{HashMap, VecDeque};
14+use std::rc::Rc;
1315
1416 use miniz_oxide::inflate::TINFLStatus;
1517 use miniz_oxide::inflate::core::{DecompressorOxide, decompress, inflate_flags};
1618 use sha1::{Digest, Sha1};
1719
18−/// Beyond this much inflated content the pack is not read: a push that
19−/// large is let through unread rather than risk the worker's memory.
20−pub const MAX_INFLATED: usize = 48 * 1024 * 1024;
20+/// Beyond this much content, counted as each object is in full (a delta
21+/// as the object it makes), the pack is not read: the work of reading
22+/// it, not the memory, which stays within [`CACHE_BYTES`] and the pack.
23+pub const MAX_INFLATED: usize = 512 * 1024 * 1024;
24+/// Objects kept inflated at a time, as bases for the deltas that follow
25+/// them. Others are inflated again from the pack when asked for.
26+const CACHE_BYTES: usize = 24 * 1024 * 1024;
2127
2228 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
2329 pub enum ObjectKind {
6773 }
6874
6975 enum Base {
70− /// An earlier object in the pack, by its offset.
71− Offset(usize),
76+ /// An earlier entry in the pack, by its index.
77+ Entry(usize),
7278 /// Any object, by id.
7379 Id(String),
7480 }
7581
76−struct Delta {
77− base: Base,
78− data: Vec<u8>,
82+enum State {
83+ /// Read, with its id known.
84+ Known(ObjectKind),
85+ /// A delta whose base is not known yet.
86+ Waiting,
87+ /// A delta that does not apply, or whose base is gone: dropped.
88+ Dropped,
7989 }
8090
81−/// The objects of a pack, by id.
91+/// An object read: its kind and its content, shared with the cache.
92+pub type Object = (ObjectKind, Rc<[u8]>);
93+
94+/// One entry of the pack, read in place.
95+struct Entry {
96+ /// Where its compressed data starts.
97+ data: usize,
98+ /// Its inflated size: for a delta, the delta's.
99+ size: usize,
100+ base: Option<Base>,
101+ state: State,
102+}
103+
104+/// Objects inflated lately, by entry, up to [`CACHE_BYTES`]; the first
105+/// kept is the first let go.
106+#[derive(Default)]
107+struct Cache {
108+ objects: HashMap<usize, Rc<[u8]>>,
109+ order: VecDeque<usize>,
110+ bytes: usize,
111+ budget: usize,
112+}
113+
114+impl Cache {
115+ fn get(&self, entry: usize) -> Option<Rc<[u8]>> {
116+ self.objects.get(&entry).cloned()
117+ }
118+
119+ fn keep(&mut self, entry: usize, data: Rc<[u8]>) {
120+ if data.len() > self.budget || self.objects.contains_key(&entry) {
121+ return;
122+ }
123+ while self.bytes + data.len() > self.budget {
124+ let Some(oldest) = self.order.pop_front() else { break };
125+ if let Some(gone) = self.objects.remove(&oldest) {
126+ self.bytes -= gone.len();
127+ }
128+ }
129+ self.bytes += data.len();
130+ self.order.push_back(entry);
131+ self.objects.insert(entry, data);
132+ }
133+}
134+
135+/// The objects of a pack, by id. The pack is kept as it came, compressed,
136+/// and an object is inflated (its deltas applied) when it is asked for:
137+/// what is held inflated at once stays within [`CACHE_BYTES`], however
138+/// much the pack holds.
82139 #[derive(Default)]
83140 pub struct Pack {
84− objects: HashMap<String, (ObjectKind, Vec<u8>)>,
85− /// Ids of the objects at each offset, once resolved.
86− at_offset: HashMap<usize, String>,
87− pending: Vec<(usize, Delta)>,
88− /// Commits, in the order the pack holds them.
141+ pack: Vec<u8>,
142+ entries: Vec<Entry>,
143+ /// Each object's entry, by id.
144+ ids: HashMap<String, usize>,
145+ /// Bases the repository supplied for a thin pack.
146+ supplied: HashMap<String, Object>,
147+ /// Commits, in the order they were read.
89148 commits: Vec<String>,
149+ cache: RefCell<Cache>,
90150 }
91151
92152 /// Where the pack starts in a receive-pack request: after the commands
240300 /// Reads every object in `pack`, resolving the deltas whose bases are
241301 /// in it.
242302 pub fn parse(pack: &[u8]) -> Result<Pack, String> {
303+ Pack::parse_within(pack, CACHE_BYTES)
304+ }
305+
306+ /// [`Pack::parse`], holding at most `cache` bytes inflated.
307+ fn parse_within(pack: &[u8], cache: usize) -> Result<Pack, String> {
243308 if pack.len() < 12 || &pack[..4] != b"PACK" {
244309 return Err("not a pack".into());
245310 }
246311 let count = u32::from_be_bytes([pack[8], pack[9], pack[10], pack[11]]) as usize;
312+ let cache = RefCell::new(Cache { budget: cache, ..Cache::default() });
313+ let mut result = Pack { pack: pack.to_vec(), cache, ..Pack::default() };
314+ let mut at_offset: HashMap<usize, usize> = HashMap::new();
247315 let mut at = 12;
248− let mut result = Pack::default();
249316 let mut inflated = 0usize;
250− for _ in 0..count {
317+ for index in 0..count {
251318 let start = at;
252319 let mut byte = *pack.get(at).ok_or("the pack ends early")?;
253320 at += 1;
260327 size |= ((byte & 0x7f) as usize) << shift;
261328 shift += 7;
262329 }
263− inflated += size;
264− if inflated > MAX_INFLATED {
265− return Err("the pack is too large to read".into());
266− }
267330 let base = match code {
268331 6 => {
269332 let mut byte = *pack.get(at).ok_or("the pack ends early")?;
274337 at += 1;
275338 offset = ((offset + 1) << 7) | (byte & 0x7f) as usize;
276339 }
277− Some(Base::Offset(start.checked_sub(offset).ok_or("a delta points before the pack")?))
340+ let base = start.checked_sub(offset).ok_or("a delta points before the pack")?;
341+ Some(Base::Entry(*at_offset.get(&base).ok_or("a delta points between objects")?))
278342 }
279343 7 => {
280344 let id = pack.get(at..at + 20).ok_or("the pack ends early")?;
283347 }
284348 _ => None,
285349 };
350+ let state = match base {
351+ Some(_) => State::Waiting,
352+ None => State::Known(ObjectKind::from_type(code).ok_or("an object has an unknown type")?),
353+ };
286354 let (data, consumed) = inflate(&pack[at..], size)?;
355+ result.entries.push(Entry { data: at, size, base, state });
356+ at_offset.insert(start, index);
287357 at += consumed;
288− match base {
289− Some(base) => result.pending.push((start, Delta { base, data })),
290− None => {
291− let kind = ObjectKind::from_type(code).ok_or("an object has an unknown type")?;
292− result.insert(start, kind, data);
358+ // Named now, while it is at hand: a whole object, or a delta
359+ // whose base is known; a delta whose base is not, once it is.
360+ inflated += match result.entries[index].state {
361+ State::Known(kind) => {
362+ let size = data.len();
363+ result.name(index, kind, data.into());
364+ size
293365 }
366+ _ => result.resolve_one(index, Some(data)).unwrap_or(0),
367+ };
368+ if inflated > MAX_INFLATED {
369+ return Err("the pack is too large to read".into());
294370 }
295371 }
296− result.resolve();
372+ if inflated + result.resolve() > MAX_INFLATED {
373+ return Err("the pack is too large to read".into());
374+ }
297375 Ok(result)
298376 }
299377
300− fn insert(&mut self, offset: usize, kind: ObjectKind, data: Vec<u8>) {
378+ /// Records `entry` as the object `data` of `kind`, kept for the deltas
379+ /// that may follow it.
380+ fn name(&mut self, entry: usize, kind: ObjectKind, data: Rc<[u8]>) {
301381 let id = object_id(kind, &data);
302382 if kind == ObjectKind::Commit {
303383 self.commits.push(id.clone());
304384 }
305− self.at_offset.insert(offset, id.clone());
306− self.objects.insert(id, (kind, data));
385+ self.entries[entry].state = State::Known(kind);
386+ self.ids.entry(id).or_insert(entry);
387+ self.cache.get_mut().keep(entry, data);
388+ }
389+
390+ /// Names the waiting delta at `entry` (inflated already when `delta`
391+ /// is given) if its base is known by now, and says how large the
392+ /// object it makes is; drops it if it does not apply. `None` while it
393+ /// waits.
394+ fn resolve_one(&mut self, entry: usize, delta: Option<Vec<u8>>) -> Option<usize> {
395+ match self.base_of(entry)? {
396+ Ok((kind, base)) => match delta.map_or_else(|| self.inflate_entry(entry), Ok).and_then(|delta| apply_delta(&base, &delta)) {
397+ Ok(data) => {
398+ let size = data.len();
399+ self.name(entry, kind, data.into());
400+ Some(size)
401+ }
402+ Err(_) => {
403+ self.entries[entry].state = State::Dropped;
404+ Some(0)
405+ }
406+ },
407+ Err(()) => {
408+ self.entries[entry].state = State::Dropped;
409+ Some(0)
410+ }
411+ }
307412 }
308413
309− /// Resolves every pending delta whose base is known by now.
310− fn resolve(&mut self) {
414+ /// The base of the delta at `entry`: `None` while it is not known,
415+ /// `Err` when it was dropped.
416+ fn base_of(&self, entry: usize) -> Option<Result<Object, ()>> {
417+ let base = match self.entries[entry].base.as_ref()? {
418+ Base::Entry(base) => *base,
419+ Base::Id(id) => match self.supplied.get(id) {
420+ Some((kind, data)) => return Some(Ok((*kind, data.clone()))),
421+ None => *self.ids.get(id)?,
422+ },
423+ };
424+ match self.entries[base].state {
425+ State::Known(_) => Some(self.object_at(base).ok_or(())),
426+ State::Waiting => None,
427+ State::Dropped => Some(Err(())),
428+ }
429+ }
430+
431+ /// Resolves every waiting delta whose base is known by now, and says
432+ /// how much they made.
433+ fn resolve(&mut self) -> usize {
434+ let mut made = 0;
311435 loop {
312436 let mut progress = false;
313− let pending = std::mem::take(&mut self.pending);
314− for (offset, delta) in pending {
315− let base_id = match &delta.base {
316− Base::Offset(base) => self.at_offset.get(base).cloned(),
317− Base::Id(id) => Some(id.clone()),
318− };
319− let resolved = base_id
320− .and_then(|id| self.objects.get(&id))
321− .map(|(kind, base)| (*kind, apply_delta(base, &delta.data)));
322− match resolved {
323− Some((kind, Ok(data))) => {
324− self.insert(offset, kind, data);
325− progress = true;
326− }
327− // A delta that does not apply is dropped.
328− Some((_, Err(_))) => progress = true,
329− None => self.pending.push((offset, delta)),
437+ for entry in 0..self.entries.len() {
438+ if matches!(self.entries[entry].state, State::Waiting)
439+ && let Some(size) = self.resolve_one(entry, None)
440+ {
441+ made += size;
442+ progress = true;
330443 }
331444 }
332− if !progress || self.pending.is_empty() {
333− return;
445+ if !progress {
446+ return made;
334447 }
335448 }
336449 }
337450
451+ /// The stored data of `entry`, inflated: for a delta, the delta.
452+ fn inflate_entry(&self, entry: usize) -> Result<Vec<u8>, String> {
453+ let entry = &self.entries[entry];
454+ inflate(&self.pack[entry.data..], entry.size).map(|(data, _)| data)
455+ }
456+
457+ /// The object at a known `entry`, from the cache or inflated again
458+ /// from the pack: down its chain of deltas to an object at hand, then
459+ /// each delta applied on the way back up.
460+ fn object_at(&self, entry: usize) -> Option<Object> {
461+ let State::Known(kind) = self.entries[entry].state else { return None };
462+ let mut chain = Vec::new();
463+ let mut at = entry;
464+ let mut data = loop {
465+ if let Some(data) = self.cache.borrow().get(at) {
466+ break data;
467+ }
468+ match self.entries[at].base.as_ref() {
469+ None => break Rc::from(self.inflate_entry(at).ok()?),
470+ Some(base) => {
471+ chain.push(at);
472+ match base {
473+ Base::Entry(base) => at = *base,
474+ Base::Id(id) => match self.supplied.get(id) {
475+ Some((_, data)) => break data.clone(),
476+ None => at = *self.ids.get(id)?,
477+ },
478+ }
479+ }
480+ }
481+ };
482+ while let Some(delta) = chain.pop() {
483+ data = apply_delta(&data, &self.inflate_entry(delta).ok()?).ok()?.into();
484+ self.cache.borrow_mut().keep(delta, data.clone());
485+ }
486+ Some((kind, data))
487+ }
488+
338489 /// Objects the pack's deltas are based on that it does not hold: what
339490 /// the repository has to supply.
340491 pub fn missing_bases(&self) -> Vec<String> {
341492 let mut ids: Vec<String> = self
342− .pending
493+ .entries
343494 .iter()
344− .filter_map(|(_, delta)| match &delta.base {
345− Base::Id(id) if !self.objects.contains_key(id) => Some(id.clone()),
495+ .filter(|entry| matches!(entry.state, State::Waiting))
496+ .filter_map(|entry| match &entry.base {
497+ Some(Base::Id(id)) if !self.contains(id) => Some(id.clone()),
346498 _ => None,
347499 })
348500 .collect();
354506 /// Supplies a base object from the repository, and resolves what
355507 /// depends on it.
356508 pub fn supply(&mut self, id: &str, kind: ObjectKind, data: Vec<u8>) {
357− self.objects.insert(id.to_owned(), (kind, data));
358− self.resolve();
509+ self.supplied.insert(id.to_owned(), (kind, data.into()));
510+ let _ = self.resolve();
359511 }
360512
361513 /// Deltas still unresolved.
362514 pub fn unresolved(&self) -> usize {
363− self.pending.len()
515+ self.entries.iter().filter(|entry| matches!(entry.state, State::Waiting)).count()
364516 }
365517
366− pub fn get(&self, id: &str) -> Option<(ObjectKind, &[u8])> {
367− self.objects.get(id).map(|(kind, data)| (*kind, data.as_slice()))
518+ pub fn get(&self, id: &str) -> Option<Object> {
519+ if let Some((kind, data)) = self.supplied.get(id) {
520+ return Some((*kind, data.clone()));
521+ }
522+ self.object_at(*self.ids.get(id)?)
368523 }
369524
370525 pub fn contains(&self, id: &str) -> bool {
371− self.objects.contains_key(id)
526+ self.ids.contains_key(id) || self.supplied.contains_key(id)
372527 }
373528
374529 /// The commits in the pack: what the push adds.
378533
379534 pub fn commit(&self, id: &str) -> Option<CommitInfo> {
380535 match self.get(id)? {
381− (ObjectKind::Commit, data) => Some(parse_commit(data)),
536+ (ObjectKind::Commit, data) => Some(parse_commit(&data)),
382537 _ => None,
383538 }
384539 }
385540
386541 pub fn tree(&self, id: &str) -> Option<Vec<TreeItem>> {
387542 match self.get(id)? {
388− (ObjectKind::Tree, data) => Some(parse_tree(data)),
543+ (ObjectKind::Tree, data) => Some(parse_tree(&data)),
389544 _ => None,
390545 }
391546 }
392547
393− pub fn blob(&self, id: &str) -> Option<&[u8]> {
548+ pub fn blob(&self, id: &str) -> Option<Rc<[u8]>> {
394549 match self.get(id)? {
395550 (ObjectKind::Blob, data) => Some(data),
396551 _ => None,
402557 /// left out). How a missing base can be asked for as what it is.
403558 pub fn named_kinds(&self) -> HashMap<String, ObjectKind> {
404559 let mut kinds = HashMap::new();
405− for (kind, data) in self.objects.values() {
560+ let read = self.ids.values().filter(|entry| matches!(self.entries[**entry].state, State::Known(ObjectKind::Commit | ObjectKind::Tree)));
561+ let objects = read.filter_map(|entry| self.object_at(*entry)).chain(self.supplied.values().cloned());
562+ for (kind, data) in objects {
563+ let data = &data[..];
406564 match kind {
407565 ObjectKind::Commit => {
408566 let commit = parse_commit(data);
581739 let pack = build_pack(&[(ObjectKind::Blob, base.clone())], &[(base_id.clone(), delta(&base, base.len(), b"second\n"))]);
582740 let parsed = Pack::parse(&pack).unwrap();
583741 let grown = b"first line\nsecond\n";
584− assert_eq!(parsed.blob(&object_id(ObjectKind::Blob, grown)), Some(&grown[..]));
742+ assert_eq!(parsed.blob(&object_id(ObjectKind::Blob, grown)).as_deref(), Some(&grown[..]));
585743 assert!(parsed.missing_bases().is_empty());
586744 }
587745
588746 #[test]
747+ fn a_chain_of_deltas_is_read_again_from_the_pack_once_let_go() {
748+ // Each version a delta on the one before; nothing stays cached, so
749+ // every read inflates the chain down to the whole object again.
750+ let mut versions = vec![b"v0\n".to_vec()];
751+ let mut deltas = Vec::new();
752+ for n in 1..6 {
753+ let before = versions.last().unwrap().clone();
754+ deltas.push((object_id(ObjectKind::Blob, &before), delta(&before, before.len(), format!("v{n}\n").as_bytes())));
755+ versions.push([before, format!("v{n}\n").into_bytes()].concat());
756+ }
757+ let pack = Pack::parse_within(&build_pack(&[(ObjectKind::Blob, versions[0].clone())], &deltas), 0).unwrap();
758+ assert_eq!(pack.unresolved(), 0);
759+ assert!(pack.cache.borrow().objects.is_empty());
760+ for version in versions.iter().rev() {
761+ assert_eq!(pack.blob(&object_id(ObjectKind::Blob, version)).as_deref(), Some(&version[..]));
762+ }
763+ }
764+
765+ #[test]
766+ fn the_cache_lets_the_oldest_go_first() {
767+ let mut cache = Cache { budget: 8, ..Cache::default() };
768+ cache.keep(0, Rc::from(&b"abcd"[..]));
769+ cache.keep(1, Rc::from(&b"efgh"[..]));
770+ cache.keep(2, Rc::from(&b"ij"[..]));
771+ assert!(cache.get(0).is_none() && cache.get(1).is_some() && cache.get(2).is_some());
772+ assert_eq!(cache.bytes, 6);
773+ // Larger than the whole budget: not kept, nothing let go for it.
774+ cache.keep(3, Rc::from(&b"0123456789"[..]));
775+ assert!(cache.get(3).is_none() && cache.get(1).is_some());
776+ }
777+
778+ #[test]
589779 fn a_thin_pack_waits_for_its_base() {
590780 let base = b"kept in the repository\n".to_vec();
591781 let base_id = object_id(ObjectKind::Blob, &base);
+0−0

Binary or large file; its contents are not shown.

+8−8
8383 if out.len() > limit {
8484 return None;
8585 }
86− queue.extend(read_commit(data).parents);
86+ queue.extend(read_commit(&data).parents);
8787 }
8888 Some(out)
8989 }
9797 let Some((ObjectKind::Commit, data)) = pack.get(id) else {
9898 return None;
9999 };
100− let commit = read_commit(data);
100+ let commit = read_commit(&data);
101101 Some(CommitFacts {
102102 sha: id.clone(),
103103 message: commit.message,
179179 let Some((ObjectKind::Commit, data)) = objects.pack.get(id) else {
180180 return Ok(None);
181181 };
182− let commit = read_commit(data);
182+ let commit = read_commit(&data);
183183 let old_tree = match commit.parents.first() {
184184 Some(parent) => objects.commit_tree(parent).await?,
185185 None => None,
215215 continue;
216216 }
217217 match pack.get(&id) {
218− Some((ObjectKind::Commit, data)) => queue.extend(read_commit(data).parents),
218+ Some((ObjectKind::Commit, data)) => queue.extend(read_commit(&data).parents),
219219 _ => boundary.push(id),
220220 }
221221 }
257257 let mut emails = Vec::new();
258258 for id in ids {
259259 let Some((ObjectKind::Commit, data)) = pack.get(id) else { continue };
260− if let Some(fingerprint) = crate::signatures::fingerprint(data)
260+ if let Some(fingerprint) = crate::signatures::fingerprint(&data)
261261 && !fingerprints.contains(&fingerprint)
262262 {
263263 fingerprints.push(fingerprint);
264− if let Some(email) = read_commit(data).committer_email.map(|email| email.to_lowercase())
264+ if let Some(email) = read_commit(&data).committer_email.map(|email| email.to_lowercase())
265265 && !emails.contains(&email)
266266 {
267267 emails.push(email);
285285 let read = try_join_all(ids.iter().map(|id| async move {
286286 let signature = owners.and_then(|(keys, emails)| {
287287 let (_, data) = pack.get(id)?;
288− let committer = read_commit(data).committer_email;
289− Some(crate::signatures::decide(data, committer.as_deref(), keys, emails))
288+ let committer = read_commit(&data).committer_email;
289+ Some(crate::signatures::decide(&data, committer.as_deref(), keys, emails))
290290 });
291291 facts(objects, id, signature).await
292292 }))
+1−1
8787 let Some((ObjectKind::Commit, data)) = pack.get(id) else {
8888 continue;
8989 };
90− let commit = rule_facts::read_commit(data);
90+ let commit = rule_facts::read_commit(&data);
9191 let old_root = match commit.parents.first() {
9292 Some(parent) => objects.commit_tree(parent).await?,
9393 None => None,

Comments

Sign in to comment.