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