593 lines · 23 KBtext/plain; charset=utf-8
Onset: version control where nobody waits and nothing gets lost. The on command, its core and onset-server.
| 6 | use std::io::Write; |
| 7 | use std::path::{Path, PathBuf}; |
| 8 | use std::time::SystemTime; |
| 9 |
| 10 | use anyhow::{Context, Result, anyhow}; |
| 11 | use onset_core::bundle::{self, Frame, RAW, ZSTD}; |
| 12 | use onset_core::object::{Kind, Object}; |
| 13 | use onset_core::{Id, Source, chunk}; |
| 14 |
| 15 | /// Objects at least this big are compressed on disk and on the wire. |
| 16 | const COMPRESS_OVER: usize = 512; |
| 17 |
| 18 | /// Up to this many objects, [`flush`] syncs each one rather than the whole disk. |
| 19 | const SYNC_EACH_UPTO: usize = 64; |
| 20 |
| 21 | /// Objects written since the last [`flush`]: on disk as far as this process |
| 22 | /// can tell, but maybe not yet past a power cut. |
| 23 | static UNFLUSHED: std::sync::Mutex<Vec<PathBuf>> = std::sync::Mutex::new(Vec::new()); |
| 24 |
| 25 | /// Makes every object written so far durable, so the database never names an |
| 26 | /// object a power cut could take back. The database calls it before it writes |
| 27 | /// anything that names objects (saves, versions, changes). Each file and its |
| 28 | /// folder; on Linux, one `syncfs` for many. A `syncfs` also waits for |
| 29 | /// everything else the disk has yet to write: after an on get of a big |
| 30 | /// project, its gigabyte of files (coline-app's first land took 7 s in it). |
| 31 | pub fn flush() -> Result<()> { |
| 32 | let pending = std::mem::take(&mut *UNFLUSHED.lock().unwrap_or_else(|e| e.into_inner())); |
| 33 | if pending.is_empty() { |
| 34 | return Ok(()); |
| 35 | } |
| 36 | #[cfg(target_os = "linux")] |
| 37 | if pending.len() > SYNC_EACH_UPTO { |
| 38 | use std::os::fd::AsRawFd; |
| 39 | if let Some(dir) = pending[0].parent() |
| 40 | && let Ok(d) = fs::File::open(dir) |
| 41 | && unsafe { libc::syncfs(d.as_raw_fd()) } == 0 |
| 42 | { |
| 43 | return Ok(()); |
| 44 | } |
| 45 | } |
| 46 | let mut dirs = std::collections::BTreeSet::new(); |
| 47 | for p in &pending { |
| 48 | // Windows flushes only a file opened for writing. A flush that fails |
| 49 | // doesn't fail the command: the object is on disk as far as anyone |
| 50 | // can tell, and on check finds and repairs one a power cut broke. |
| 51 | if let Ok(f) = fs::OpenOptions::new().write(true).open(p) { |
| 52 | let _ = f.sync_data(); |
| 53 | } |
| 54 | if let Some(d) = p.parent() { |
| 55 | dirs.insert(d.to_path_buf()); |
| 56 | } |
| 57 | } |
| 58 | #[cfg(unix)] |
| 59 | for d in dirs { |
| 60 | if let Ok(f) = fs::File::open(&d) { |
| 61 | let _ = f.sync_all(); |
| 62 | } |
| 63 | } |
| 64 | Ok(()) |
| 65 | } |
| 66 |
| 67 | /// Objects are written one file each; `on gc` moves older ones into a pack |
| 68 | /// (one file holding many, as they were) with a sorted index, so a big |
| 69 | /// project isn't tens of thousands of small files. Reads look in both. |
| 70 | const PACKS: &str = "packs"; |
| 71 | const IDX_MAGIC: &[u8; 8] = b"ONSETIX1"; |
| 72 | /// A bundle bringing at least this many new objects, and at most this many |
| 73 | /// bytes of them, is stored as a pack. |
| 74 | const PACK_FROM: usize = 256; |
| 75 | const PACK_UPTO: usize = 64 << 20; |
| 76 |
| 77 | /// An object as its file holds it (and a pack): kind, codec, size, then the payload. |
| 78 | fn stored(frame: &Frame) -> Vec<u8> { |
| 79 | let mut bytes = Vec::with_capacity(frame.payload.len() + 10); |
| 80 | bytes.push(frame.kind as u8); |
| 81 | bytes.push(frame.codec); |
| 82 | bytes.extend_from_slice(&frame.raw_len.to_le_bytes()); |
| 83 | bytes.extend_from_slice(&frame.payload); |
| 84 | bytes |
| 85 | } |
| 86 |
| 87 | /// A pack's index: (ID, offset, length) sorted by ID. |
| 88 | struct PackIndex { |
| 89 | pack: PathBuf, |
| 90 | entries: Vec<(Id, u64, u64)>, |
| 91 | } |
| 92 |
| 93 | #[derive(Default)] |
| 94 | struct Packs { |
| 95 | list: Vec<PackIndex>, |
| 96 | /// The packs folder's modified time when read: another process's pack shows as a change. |
| 97 | seen: Option<SystemTime>, |
| 98 | } |
| 99 |
| 100 | #[derive(Clone)] |
| 101 | pub struct Store { |
| 102 | root: PathBuf, |
| 103 | packs: std::sync::Arc<std::sync::RwLock<Packs>>, |
| 104 | } |
| 105 |
| 106 | impl Store { |
| 107 | pub fn open(root: &Path) -> Result<Store> { |
| 108 | fs::create_dir_all(root).with_context(|| format!("creating {}", root.display()))?; |
| 109 | Ok(Store { root: root.to_path_buf(), packs: Default::default() }) |
| 110 | } |
| 111 |
| 112 | fn path(&self, id: &Id) -> PathBuf { |
| 113 | let hex = id.hex(); |
| 114 | self.root.join(&hex[..2]).join(&hex[2..]) |
| 115 | } |
| 116 |
| 117 | pub fn has(&self, id: &Id) -> bool { |
| 118 | self.path(id).is_file() || self.packed(id).is_some() |
| 119 | } |
| 120 |
| 121 | /// Reads the packs' indexes again if the packs folder changed since. |
| 122 | fn load_packs(&self, force: bool) { |
| 123 | let dir = self.root.join(PACKS); |
| 124 | let modified = fs::metadata(&dir).and_then(|m| m.modified()).ok(); |
| 125 | if !force && self.packs.read().unwrap_or_else(|e| e.into_inner()).seen == modified { |
| 126 | return; |
| 127 | } |
| 128 | let mut list = Vec::new(); |
| 129 | for f in fs::read_dir(&dir).into_iter().flatten().flatten() { |
| 130 | let p = f.path(); |
| 131 | if p.extension().and_then(|e| e.to_str()) != Some("idx") { |
| 132 | continue; |
| 133 | } |
| 134 | let Ok(bytes) = fs::read(&p) else { continue }; |
| 135 | if bytes.len() < 16 || &bytes[..8] != IDX_MAGIC { |
| 136 | continue; |
| 137 | } |
| 138 | let n = u64::from_le_bytes(bytes[8..16].try_into().unwrap()) as usize; |
| 139 | if bytes.len() != 16 + n * 48 { |
| 140 | continue; |
| 141 | } |
| 142 | let entries = (0..n) |
| 143 | .map(|i| { |
| 144 | let at = 16 + i * 48; |
| 145 | let id = Id(bytes[at..at + 32].try_into().unwrap()); |
| 146 | (id, u64::from_le_bytes(bytes[at + 32..at + 40].try_into().unwrap()), u64::from_le_bytes(bytes[at + 40..at + 48].try_into().unwrap())) |
| 147 | }) |
| 148 | .collect(); |
| 149 | list.push(PackIndex { pack: p.with_extension("pack"), entries }); |
| 150 | } |
| 151 | *self.packs.write().unwrap_or_else(|e| e.into_inner()) = Packs { list, seen: modified }; |
| 152 | } |
| 153 |
| 154 | /// Where a packed object is: (pack, offset, length). |
| 155 | fn packed(&self, id: &Id) -> Option<(PathBuf, u64, u64)> { |
| 156 | self.load_packs(false); |
| 157 | let packs = self.packs.read().unwrap_or_else(|e| e.into_inner()); |
| 158 | packs.list.iter().find_map(|p| p.entries.binary_search_by(|e| e.0.cmp(id)).ok().map(|i| (p.pack.clone(), p.entries[i].1, p.entries[i].2))) |
| 159 | } |
| 160 |
| 161 | fn read_packed(&self, id: &Id) -> Result<Option<Vec<u8>>> { |
| 162 | use std::io::{Read, Seek, SeekFrom}; |
| 163 | for attempt in 0..2 { |
| 164 | let Some((pack, off, len)) = self.packed(id) else { return Ok(None) }; |
| 165 | let read = (|| -> std::io::Result<Vec<u8>> { |
| 166 | let mut f = fs::File::open(&pack)?; |
| 167 | f.seek(SeekFrom::Start(off))?; |
| 168 | let mut buf = vec![0; len as usize]; |
| 169 | f.read_exact(&mut buf)?; |
| 170 | Ok(buf) |
| 171 | })(); |
| 172 | match read { |
| 173 | Ok(buf) => return Ok(Some(buf)), |
| 174 | // Repacked by another process meanwhile: read the indexes again. |
| 175 | Err(_) if attempt == 0 => self.load_packs(true), |
| 176 | Err(e) => return Err(e.into()), |
| 177 | } |
| 178 | } |
| 179 | Ok(None) |
| 180 | } |
| 181 |
| 182 | /// Every packed object's ID. |
| 183 | pub fn packed_ids(&self) -> Vec<Id> { |
| 184 | self.load_packs(false); |
| 185 | self.packs.read().unwrap_or_else(|e| e.into_inner()).list.iter().flat_map(|p| p.entries.iter().map(|e| e.0)).collect() |
| 186 | } |
| 187 |
| 188 | /// Moves loose objects not modified for `older` into a new pack. |
| 189 | /// Answers how many it moved. |
| 190 | pub fn pack_loose(&self, older: std::time::Duration) -> Result<usize> { |
| 191 | let now = SystemTime::now(); |
| 192 | let mut picked: Vec<(Id, PathBuf)> = self |
| 193 | .all()? |
| 194 | .into_iter() |
| 195 | .filter(|(_, p)| fs::metadata(p).and_then(|m| m.modified()).ok().and_then(|m| now.duration_since(m).ok()).is_some_and(|age| age >= older)) |
| 196 | .collect(); |
| 197 | if picked.is_empty() { |
| 198 | return Ok(0); |
| 199 | } |
| 200 | picked.sort_by(|a, b| a.0.cmp(&b.0)); |
| 201 | let entries: Vec<(Id, Vec<u8>)> = picked.iter().filter_map(|(id, p)| fs::read(p).ok().map(|b| (*id, b))).collect(); |
| 202 | self.write_pack(&entries)?; |
| 203 | for (_, p) in &picked { |
| 204 | let _ = fs::remove_file(p); |
| 205 | } |
| 206 | Ok(entries.len()) |
| 207 | } |
| 208 |
| 209 | /// Writes a pack and its index (the index last: an index means a whole pack). |
| 210 | fn write_pack(&self, entries: &[(Id, Vec<u8>)]) -> Result<()> { |
| 211 | let dir = self.root.join(PACKS); |
| 212 | fs::create_dir_all(&dir)?; |
| 213 | let name = format!("{:016x}{:08x}", rand::random::<u64>(), std::process::id()); |
| 214 | let (pack, idx) = (dir.join(format!("{name}.pack")), dir.join(format!("{name}.idx"))); |
| 215 | let mut body = Vec::new(); |
| 216 | let mut index = Vec::with_capacity(16 + entries.len() * 48); |
| 217 | index.extend_from_slice(IDX_MAGIC); |
| 218 | index.extend_from_slice(&(entries.len() as u64).to_le_bytes()); |
| 219 | for (id, bytes) in entries { |
| 220 | index.extend_from_slice(&id.0); |
| 221 | index.extend_from_slice(&(body.len() as u64).to_le_bytes()); |
| 222 | index.extend_from_slice(&(bytes.len() as u64).to_le_bytes()); |
| 223 | body.extend_from_slice(bytes); |
| 224 | } |
| 225 | for (path, data) in [(&pack, &body), (&idx, &index)] { |
| 226 | let tmp = path.with_extension("tmp"); |
| 227 | let mut f = fs::File::create(&tmp)?; |
| 228 | f.write_all(data)?; |
| 229 | f.sync_all()?; |
| 230 | fs::rename(&tmp, path)?; |
| 231 | } |
| 232 | #[cfg(unix)] |
| 233 | if let Ok(d) = fs::File::open(&dir) { |
| 234 | let _ = d.sync_all(); |
| 235 | } |
| 236 | self.load_packs(true); |
| 237 | Ok(()) |
| 238 | } |
| 239 |
| 240 | /// Takes objects out of the packs that hold them (each such pack is |
| 241 | /// written again without them, so their bytes are gone). Answers the bytes freed. |
| 242 | pub fn remove_packed(&self, ids: &std::collections::HashSet<Id>) -> Result<u64> { |
| 243 | self.load_packs(true); |
| 244 | let packs: Vec<(PathBuf, Vec<(Id, u64, u64)>)> = self.packs.read().unwrap_or_else(|e| e.into_inner()).list.iter().map(|p| (p.pack.clone(), p.entries.clone())).collect(); |
| 245 | let mut freed = 0; |
| 246 | for (pack, entries) in packs { |
| 247 | if !entries.iter().any(|e| ids.contains(&e.0)) { |
| 248 | continue; |
| 249 | } |
| 250 | let data = fs::read(&pack)?; |
| 251 | let keep: Vec<(Id, Vec<u8>)> = entries.iter().filter(|e| !ids.contains(&e.0)).map(|e| (e.0, data[e.1 as usize..(e.1 + e.2) as usize].to_vec())).collect(); |
| 252 | freed += entries.iter().filter(|e| ids.contains(&e.0)).map(|e| e.2).sum::<u64>(); |
| 253 | if !keep.is_empty() { |
| 254 | self.write_pack(&keep)?; |
| 255 | } |
| 256 | fs::remove_file(pack.with_extension("idx"))?; |
| 257 | fs::remove_file(&pack)?; |
| 258 | } |
| 259 | self.load_packs(true); |
| 260 | Ok(freed) |
| 261 | } |
| 262 |
| 263 | /// Every object here, with its file. |
| 264 | pub fn all(&self) -> Result<Vec<(Id, PathBuf)>> { |
| 265 | let mut out = Vec::new(); |
| 266 | for dir in fs::read_dir(&self.root)?.flatten() { |
| 267 | let prefix = dir.file_name().to_string_lossy().to_string(); |
| 268 | if prefix.len() != 2 || !dir.path().is_dir() { |
| 269 | continue; |
| 270 | } |
| 271 | for file in fs::read_dir(dir.path())?.flatten() { |
| 272 | if let Some(id) = Id::parse(&format!("{prefix}{}", file.file_name().to_string_lossy())) { |
| 273 | out.push((id, file.path())); |
| 274 | } |
| 275 | } |
| 276 | } |
| 277 | Ok(out) |
| 278 | } |
| 279 |
| 280 | /// Deletes an object (on purge); answers the bytes it took. A packed |
| 281 | /// one's pack is written again without it, so its bytes are gone too. |
| 282 | pub fn remove(&self, id: &Id) -> u64 { |
| 283 | let loose = fs::metadata(self.path(id)).map(|m| m.len()).ok(); |
| 284 | let mut freed = 0; |
| 285 | if let Some(len) = loose |
| 286 | && fs::remove_file(self.path(id)).is_ok() |
| 287 | { |
| 288 | freed += len; |
| 289 | } |
| 290 | if self.packed(id).is_some() { |
| 291 | freed += self.remove_packed(&std::iter::once(*id).collect()).unwrap_or(0); |
| 292 | } |
| 293 | freed |
| 294 | } |
| 295 |
| 296 | /// How many bytes the object takes here (0 when it's missing). |
| 297 | pub fn stored_len(&self, id: &Id) -> u64 { |
| 298 | fs::metadata(self.path(id)).map(|m| m.len()).ok().or_else(|| self.packed(id).map(|p| p.2)).unwrap_or(0) |
| 299 | } |
| 300 |
| 301 | /// Stores an object (no-op if present). Returns its ID. |
| 302 | pub fn put(&self, object: &Object) -> Result<Id> { |
| 303 | let id = object.id(); |
| 304 | if self.has(&id) { |
| 305 | return Ok(id); |
| 306 | } |
| 307 | self.write_frame(&encode(object, id))?; |
| 308 | Ok(id) |
| 309 | } |
| 310 |
| 311 | fn write_frame(&self, frame: &Frame) -> Result<()> { |
| 312 | let path = self.path(&frame.id); |
| 313 | if path.is_file() || self.packed(&frame.id).is_some() { |
| 314 | return Ok(()); |
| 315 | } |
| 316 | let dir = path.parent().unwrap(); |
| 317 | fs::create_dir_all(dir)?; |
| 318 | let bytes = stored(frame); |
| 319 | let tmp = dir.join(format!(".tmp-{}-{}", std::process::id(), rand::random::<u32>())); |
| 320 | { |
| 321 | let mut f = fs::File::create(&tmp)?; |
| 322 | f.write_all(&bytes)?; |
| 323 | } |
| 324 | fs::rename(&tmp, &path)?; |
| 325 | UNFLUSHED.lock().unwrap_or_else(|e| e.into_inner()).push(path); |
| 326 | Ok(()) |
| 327 | } |
| 328 |
| 329 | /// Deletes a damaged object, so it can be fetched again. |
| 330 | pub fn forget(&self, id: &Id) { |
| 331 | let _ = fs::remove_file(self.path(id)); |
| 332 | if self.packed(id).is_some() { |
| 333 | let _ = self.remove_packed(&std::iter::once(*id).collect()); |
| 334 | } |
| 335 | } |
| 336 |
| 337 | /// The stored frame (possibly compressed), for bundles. |
| 338 | pub fn frame(&self, id: &Id) -> Result<Option<Frame>> { |
| 339 | let bytes = match fs::read(self.path(id)) { |
| 340 | Ok(b) => b, |
| 341 | Err(e) if e.kind() == std::io::ErrorKind::NotFound => match self.read_packed(id)? { |
| 342 | Some(b) => b, |
| 343 | None => return Ok(None), |
| 344 | }, |
| 345 | Err(e) => return Err(e.into()), |
| 346 | }; |
| 347 | if bytes.len() < 10 { |
| 348 | return Err(anyhow!("object {} is damaged", id.short())); |
| 349 | } |
| 350 | let kind = Kind::from_u8(bytes[0]).ok_or_else(|| anyhow!("object {} is damaged", id.short()))?; |
| 351 | let codec = bytes[1]; |
| 352 | let raw_len = u64::from_le_bytes(bytes[2..10].try_into().unwrap()); |
| 353 | Ok(Some(Frame { id: *id, kind, codec, raw_len, payload: bytes[10..].to_vec() })) |
| 354 | } |
| 355 |
| 356 | pub fn read(&self, id: &Id) -> Result<Option<Object>> { |
| 357 | match self.frame(id)? { |
| 358 | Some(frame) => Ok(Some(decode(&frame)?)), |
| 359 | None => Ok(None), |
| 360 | } |
| 361 | } |
| 362 |
| 363 | /// Reads every object kept here and checks it against its ID, in |
| 364 | /// parallel. Answers how many there are and the IDs of damaged ones. |
| 365 | pub fn verify_all(&self) -> Result<(usize, Vec<Id>)> { |
| 366 | use rayon::prelude::*; |
| 367 | let mut ids = Vec::new(); |
| 368 | for dir in fs::read_dir(&self.root)? { |
| 369 | let dir = dir?; |
| 370 | let Some(prefix) = dir.file_name().to_str().filter(|p| p.len() == 2).map(str::to_string) else { continue }; |
| 371 | if !dir.file_type()?.is_dir() { |
| 372 | continue; |
| 373 | } |
| 374 | for f in fs::read_dir(dir.path())? { |
| 375 | let name = f?.file_name(); |
| 376 | if let Some(id) = name.to_str().and_then(|rest| Id::parse(&format!("{prefix}{rest}"))) { |
| 377 | ids.push(id); |
| 378 | } |
| 379 | } |
| 380 | } |
| 381 | ids.extend(self.packed_ids()); |
| 382 | let damaged: Vec<Id> = ids |
| 383 | .par_iter() |
| 384 | .filter(|id| match self.frame(id) { |
| 385 | Ok(Some(frame)) => decode(&frame).map_or(true, |o| bundle::verify(&frame, &o.data).is_err()), |
| 386 | _ => true, |
| 387 | }) |
| 388 | .copied() |
| 389 | .collect(); |
| 390 | Ok((ids.len(), damaged)) |
| 391 | } |
| 392 |
| 393 | /// Stores a file's content (chunked when big) and returns its ID. |
| 394 | pub fn put_file(&self, content: &[u8]) -> Result<Id> { |
| 395 | let objects = chunk::file_objects(content); |
| 396 | let id = objects[0].id(); |
| 397 | if self.has(&id) { |
| 398 | return Ok(id); |
| 399 | } |
| 400 | // Chunks first, so the list never points at missing chunks. |
| 401 | for o in objects.iter().skip(1) { |
| 402 | self.put(o)?; |
| 403 | } |
| 404 | self.put(&objects[0])?; |
| 405 | Ok(id) |
| 406 | } |
| 407 |
| 408 | /// A file's content: a sealed folder's file opened when this machine has its key. |
| 409 | pub fn read_file(&self, id: &Id) -> Result<Vec<u8>> { |
| 410 | Ok(crate::seal::open(chunk::read_file(self, id)?)) |
| 411 | } |
| 412 |
| 413 | /// Every object a file ID needs (itself plus chunks). |
| 414 | /// A file's size (a big file's whole size), read from its frame without |
| 415 | /// decompressing a blob; None when it isn't here. |
| 416 | pub fn file_size(&self, id: &Id) -> Result<Option<u64>> { |
| 417 | let Some(frame) = self.frame(id)? else { return Ok(None) }; |
| 418 | if frame.kind != Kind::Chunks { |
| 419 | return Ok(Some(frame.raw_len)); |
| 420 | } |
| 421 | Ok(Some(onset_core::object::decode_chunks(&decode(&frame)?.data)?.iter().map(|(_, n)| n).sum())) |
| 422 | } |
| 423 |
| 424 | pub fn file_parts(&self, id: &Id) -> Result<Vec<Id>> { |
| 425 | let mut out = vec![*id]; |
| 426 | // The first byte says whether it's a chunk list; blobs aren't read. |
| 427 | let mut kind = [0u8; 1]; |
| 428 | match fs::File::open(self.path(id)).and_then(|mut f| std::io::Read::read_exact(&mut f, &mut kind)) { |
| 429 | Ok(()) if kind[0] == Kind::Chunks as u8 => {} |
| 430 | Ok(()) => return Ok(out), |
| 431 | // Not loose: packed (read whole below), or not here. |
| 432 | Err(e) if e.kind() == std::io::ErrorKind::NotFound && self.packed(id).is_some() => {} |
| 433 | Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(out), |
| 434 | Err(e) => return Err(e.into()), |
| 435 | } |
| 436 | if let Some(frame) = self.frame(id)? |
| 437 | && frame.kind == Kind::Chunks |
| 438 | { |
| 439 | let o = decode(&frame)?; |
| 440 | out.extend(onset_core::object::decode_chunks(&o.data)?.into_iter().map(|(c, _)| c)); |
| 441 | } |
| 442 | Ok(out) |
| 443 | } |
| 444 |
| 445 | /// A bundle of the given objects (those present). |
| 446 | pub fn bundle(&self, ids: &[Id]) -> Result<Vec<u8>> { |
| 447 | let mut out = Vec::new(); |
| 448 | bundle::write_header(&mut out); |
| 449 | for id in ids { |
| 450 | if let Some(frame) = self.frame(id)? { |
| 451 | bundle::write_frame(&mut out, &frame); |
| 452 | } |
| 453 | } |
| 454 | bundle::write_end(&mut out); |
| 455 | Ok(out) |
| 456 | } |
| 457 |
| 458 | /// `bundle`, with each object that replaced one the host has (`bases`: |
| 459 | /// object → the one it replaced) as a delta on it when that's under half |
| 460 | /// its size: a folder with one entry changed goes as tens of bytes. |
| 461 | pub fn bundle_based(&self, ids: &[Id], bases: &std::collections::HashMap<Id, Id>) -> Result<Vec<u8>> { |
| 462 | let mut out = Vec::new(); |
| 463 | bundle::write_header(&mut out); |
| 464 | for id in ids { |
| 465 | let delta = match bases.get(id) { |
| 466 | Some(base) => match (self.get(id)?, self.get(base)?) { |
| 467 | (Some(target), Some(was)) if target.kind == was.kind && target.kind != onset_core::Kind::Chunks => { |
| 468 | onset_core::delta::encode(&was.data, &target.data).map(|d| bundle::Frame { id: *id, kind: target.kind, codec: bundle::DELTA, raw_len: target.data.len() as u64, payload: [&base.0[..], &d[..]].concat() }) |
| 469 | } |
| 470 | _ => None, |
| 471 | }, |
| 472 | None => None, |
| 473 | }; |
| 474 | if let Some(frame) = delta { |
| 475 | bundle::write_frame(&mut out, &frame); |
| 476 | } else if let Some(frame) = self.frame(id)? { |
| 477 | bundle::write_frame(&mut out, &frame); |
| 478 | } |
| 479 | } |
| 480 | bundle::write_end(&mut out); |
| 481 | Ok(out) |
| 482 | } |
| 483 |
| 484 | /// Stores every object in a bundle after checking it. Returns how many. |
| 485 | /// Many new ones (a project's first get) go into one pack, not a file |
| 486 | /// each: the files were half of what a get wrote. |
| 487 | pub fn ingest(&self, bytes: &[u8]) -> Result<usize> { |
| 488 | let frames = bundle::read(bytes)?; |
| 489 | let n = frames.len(); |
| 490 | let mut new = Vec::new(); |
| 491 | for frame in frames { |
| 492 | if self.has(&frame.id) { |
| 493 | continue; |
| 494 | } |
| 495 | let object = decode(&frame)?; |
| 496 | bundle::verify(&frame, &object.data)?; |
| 497 | new.push(frame); |
| 498 | } |
| 499 | // A big bundle (a big project's files) stays a file each: written as |
| 500 | // one pack, coline-app's gigabyte took longer to get, not shorter. |
| 501 | let bytes: usize = new.iter().map(|f| f.payload.len()).sum(); |
| 502 | if new.len() < PACK_FROM || bytes > PACK_UPTO { |
| 503 | for frame in &new { |
| 504 | self.write_frame(frame)?; |
| 505 | } |
| 506 | return Ok(n); |
| 507 | } |
| 508 | let mut entries: Vec<(Id, Vec<u8>)> = new.iter().map(|f| (f.id, stored(f))).collect(); |
| 509 | entries.sort_by(|a, b| a.0.cmp(&b.0)); |
| 510 | entries.dedup_by(|a, b| a.0 == b.0); |
| 511 | self.write_pack(&entries)?; |
| 512 | Ok(n) |
| 513 | } |
| 514 |
| 515 | /// `ingest` for an upload, which may hold deltas on objects stored here |
| 516 | /// or earlier in it (see `bundle_based`): each is made whole, checked |
| 517 | /// against its ID and stored whole. Together they make at most 64 MiB. |
| 518 | pub fn ingest_uploaded(&self, bytes: &[u8]) -> Result<usize> { |
| 519 | let frames = bundle::read_stored(bytes)?; |
| 520 | let n = frames.len(); |
| 521 | let mut made = 0usize; |
| 522 | for frame in frames { |
| 523 | if self.has(&frame.id) { |
| 524 | continue; |
| 525 | } |
| 526 | let object = if frame.codec == bundle::DELTA { |
| 527 | let base_id = Id::from_slice(&frame.payload[..32]).ok_or_else(|| anyhow::anyhow!("bad delta"))?; |
| 528 | let base = self.read(&base_id)?.ok_or_else(|| anyhow::anyhow!("object {} is a delta on {}, which isn't here; send it whole", frame.id.short(), base_id.short()))?; |
| 529 | made += frame.raw_len as usize; |
| 530 | if made > 64 << 20 { |
| 531 | anyhow::bail!("an upload's deltas make at most 64 MiB; send the objects whole"); |
| 532 | } |
| 533 | Object::new(frame.kind, onset_core::git::apply_delta(&base.data, &frame.payload[32..], frame.raw_len as usize)?) |
| 534 | } else { |
| 535 | decode(&frame)? |
| 536 | }; |
| 537 | bundle::verify(&frame, &object.data)?; |
| 538 | self.write_frame(&encode(&object, frame.id))?; |
| 539 | } |
| 540 | Ok(n) |
| 541 | } |
| 542 | } |
| 543 |
| 544 | impl Source for Store { |
| 545 | fn get(&self, id: &Id) -> onset_core::Result<Option<Object>> { |
| 546 | self.read(id).map_err(|e| onset_core::Error::Corrupt(format!("{e:#}"))) |
| 547 | } |
| 548 | } |
| 549 |
| 550 | pub fn encode(object: &Object, id: Id) -> Frame { |
| 551 | if object.data.len() >= COMPRESS_OVER |
| 552 | && let Ok(packed) = zstd::bulk::compress(&object.data, 3) |
| 553 | && packed.len() < object.data.len() |
| 554 | { |
| 555 | return Frame { id, kind: object.kind, codec: ZSTD, raw_len: object.data.len() as u64, payload: packed }; |
| 556 | } |
| 557 | Frame { id, kind: object.kind, codec: RAW, raw_len: object.data.len() as u64, payload: object.data.clone() } |
| 558 | } |
| 559 |
| 560 | pub fn decode(frame: &Frame) -> Result<Object> { |
| 561 | let data = match frame.codec { |
| 562 | RAW => frame.payload.clone(), |
| 563 | ZSTD => { |
| 564 | // A frame's claimed size isn't trusted: reading stops just past it. |
| 565 | use std::io::Read; |
| 566 | let mut out = Vec::with_capacity((frame.raw_len as usize).min(frame.payload.len().saturating_mul(4)).min(1 << 20)); |
| 567 | zstd::stream::read::Decoder::new(&frame.payload[..])?.take(frame.raw_len + 1).read_to_end(&mut out).context("damaged object")?; |
| 568 | out |
| 569 | } |
| 570 | _ => return Err(anyhow!("unknown codec")), |
| 571 | }; |
| 572 | Ok(Object::new(frame.kind, data)) |
| 573 | } |
| 574 |
| 575 | #[cfg(test)] |
| 576 | mod tests { |
| 577 | use super::*; |
| 578 |
| 579 | #[test] |
| 580 | fn put_read_bundle() { |
| 581 | let dir = tempfile::tempdir().unwrap(); |
| 582 | let store = Store::open(dir.path()).unwrap(); |
| 583 | let big = "x".repeat(10_000).into_bytes(); |
| 584 | let id = store.put_file(&big).unwrap(); |
| 585 | assert_eq!(store.read_file(&id).unwrap(), big); |
| 586 | let bundle = store.bundle(&[id]).unwrap(); |
| 587 | let other = tempfile::tempdir().unwrap(); |
| 588 | let store2 = Store::open(other.path()).unwrap(); |
| 589 | assert_eq!(store2.ingest(&bundle).unwrap(), 1); |
| 590 | assert_eq!(store2.read_file(&id).unwrap(), big); |
| 591 | } |
| 592 | } |
| 593 |