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