Skip to content

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

Commit

A pushed pack is read in place, so one that inflates past the Worker's memory is scanned instead of answering 500 (Closes #176)

The reader inflated every object of a push and held them all, and its 48 MB cap counted a delta at the delta's size, not the object it makes. A 9.8 MB push of 359 rewritten commits inflated to 200 MB, past the Worker's 128 MB, and Cloudflare ended the request with a bare 500. Now the pack stays compressed, indexed by entry, and an object is inflated when asked for: down its chain of deltas to one at hand, then back up. The objects inflated lately are kept for the deltas that follow, up to 24 MB. That push reads in 44 MB at its peak, all 8070 objects matching git's index. The cap counts what each object is in full, and bounds the work: 512 MB.

G1t-Session:
ses_01m5b7dq37eb8tm6hbrkw3rt12
syntaqxcommitted Parentfc98f07Browse 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.