| 1 |
|
| 2 |
|
| 3 |
|
| 4 |
|
| 5 |
|
| 6 |
|
| 7 |
|
| 8 |
|
| 9 |
|
| 10 |
|
| 11 |
|
| 12 |
|
| 13 |
|
| 14 |
|
| 15 |
|
| 16 |
|
| 17 |
|
| 18 |
|
| 19 |
|
| 20 |
|
| 21 |
|
| 22 |
|
| 23 |
|
| 24 |
|
| 25 |
|
| 26 |
|
| 27 |
|
| 28 |
|
| 29 |
|
| 30 |
use crate::attest::Attest; |
| 31 |
use crate::hash::{hash_object, header, ObjectRef, ObjectType}; |
| 32 |
use crate::identity::{Identity, Signature}; |
| 33 |
use crate::objects::{Blob, Change, ObjectError, StateObject}; |
| 34 |
use crate::receipt::Receipt; |
| 35 |
use crate::strata::{self, Error as BackendError, LocalStore, MemoryStore}; |
| 36 |
use rand::RngCore; |
| 37 |
use std::cell::RefCell; |
| 38 |
use std::fs; |
| 39 |
use std::io::{self, Write}; |
| 40 |
use std::path::{Path, PathBuf}; |
| 41 |
use std::sync::Arc; |
| 42 |
|
| 43 |
const PACK_MAGIC: &[u8; 8] = b"OXYPACK1"; |
| 44 |
const IDX_MAGIC: &[u8; 8] = b"OXYIDX01"; |
| 45 |
const IDX_ENTRY: usize = 32 + 8 + 4 + 4 + 1 + 1; |
| 46 |
const FLAG_ZSTD: u8 = 1; |
| 47 |
|
| 48 |
|
| 49 |
|
| 50 |
const FLAG_DICT: u8 = 2; |
| 51 |
|
| 52 |
|
| 53 |
const DICT_BYTES: usize = 110 << 10; |
| 54 |
const DICT_MIN_OBJECTS: usize = 1000; |
| 55 |
|
| 56 |
const DICT_OBJECT_MAX: usize = 64 << 10; |
| 57 |
const ZSTD_LEVEL: i32 = 3; |
| 58 |
|
| 59 |
pub const REPACK_ZSTD_LEVEL: i32 = 9; |
| 60 |
|
| 61 |
|
| 62 |
const MAX_OBJECT_LEN: u32 = 1 << 30; |
| 63 |
|
| 64 |
|
| 65 |
|
| 66 |
const NO_DICT: &str = "packed with a dictionary the pack lacks"; |
| 67 |
|
| 68 |
|
| 69 |
|
| 70 |
|
| 71 |
|
| 72 |
|
| 73 |
|
| 74 |
|
| 75 |
|
| 76 |
const RETIRED_KEY: &str = "pack-retired"; |
| 77 |
|
| 78 |
|
| 79 |
|
| 80 |
|
| 81 |
pub const RETIRED_FOR: std::time::Duration = std::time::Duration::from_secs(60 * 60); |
| 82 |
|
| 83 |
const RETIRED_TRIES: usize = 32; |
| 84 |
|
| 85 |
|
| 86 |
|
| 87 |
|
| 88 |
const HAS_RELIST_AFTER: std::time::Duration = std::time::Duration::from_secs(1); |
| 89 |
|
| 90 |
|
| 91 |
type Listed = Vec<(String, u64, Option<u64>)>; |
| 92 |
|
| 93 |
|
| 94 |
fn parse_retired(bytes: &[u8]) -> std::collections::BTreeMap<String, u64> { |
| 95 |
String::from_utf8_lossy(bytes) |
| 96 |
.lines() |
| 97 |
.filter_map(|l| { |
| 98 |
let (t, k) = l.split_once(' ')?; |
| 99 |
Some((k.to_string(), t.parse().ok()?)) |
| 100 |
}) |
| 101 |
.collect() |
| 102 |
} |
| 103 |
|
| 104 |
fn encode_retired(m: &std::collections::BTreeMap<String, u64>) -> Vec<u8> { |
| 105 |
m.iter() |
| 106 |
.map(|(k, t)| format!("{t} {k}\n")) |
| 107 |
.collect::<String>() |
| 108 |
.into_bytes() |
| 109 |
} |
| 110 |
|
| 111 |
|
| 112 |
#[derive(Debug, Clone, Default, PartialEq, Eq)] |
| 113 |
pub struct RetiredReport { |
| 114 |
|
| 115 |
pub packs: usize, |
| 116 |
|
| 117 |
pub loose: usize, |
| 118 |
|
| 119 |
|
| 120 |
pub kept: usize, |
| 121 |
} |
| 122 |
|
| 123 |
|
| 124 |
pub type Object = (ObjectType, Vec<u8>); |
| 125 |
|
| 126 |
|
| 127 |
|
| 128 |
pub type Progress<'a> = &'a mut dyn FnMut(&str); |
| 129 |
|
| 130 |
|
| 131 |
pub type Backend = dyn strata::Store; |
| 132 |
|
| 133 |
fn silent() -> impl FnMut(&str) { |
| 134 |
|_| {} |
| 135 |
} |
| 136 |
|
| 137 |
|
| 138 |
|
| 139 |
fn backend_err(e: BackendError) -> StoreError { |
| 140 |
match e { |
| 141 |
BackendError::Io(e) => StoreError::Io(e), |
| 142 |
other => StoreError::Io(io::Error::other(other.to_string())), |
| 143 |
} |
| 144 |
} |
| 145 |
|
| 146 |
|
| 147 |
fn put_new_or_existing(backend: &Backend, key: &str, bytes: &[u8]) -> Result<()> { |
| 148 |
match backend.put_new(key, bytes) { |
| 149 |
Ok(()) | Err(BackendError::Exists(_)) => Ok(()), |
| 150 |
Err(e) => Err(backend_err(e)), |
| 151 |
} |
| 152 |
} |
| 153 |
|
| 154 |
|
| 155 |
fn get_optional(backend: &Backend, key: &str) -> Result<Option<Vec<u8>>> { |
| 156 |
match backend.get(key) { |
| 157 |
Ok(b) => Ok(Some(b)), |
| 158 |
Err(BackendError::NotFound(_)) => Ok(None), |
| 159 |
Err(e) => Err(backend_err(e)), |
| 160 |
} |
| 161 |
} |
| 162 |
|
| 163 |
fn delete(backend: &Backend, key: &str) -> Result<()> { |
| 164 |
backend.delete(key).map_err(backend_err) |
| 165 |
} |
| 166 |
|
| 167 |
|
| 168 |
fn loose_key(r: &ObjectRef) -> String { |
| 169 |
let h = r.to_hex(); |
| 170 |
format!("loose/{}/{}", &h[..2], &h[2..]) |
| 171 |
} |
| 172 |
|
| 173 |
|
| 174 |
fn loose_ref(key: &str) -> Option<ObjectRef> { |
| 175 |
let rest = key.strip_prefix("loose/")?; |
| 176 |
let (hh, tail) = rest.split_once('/')?; |
| 177 |
if hh.len() != 2 || tail.contains('/') { |
| 178 |
return None; |
| 179 |
} |
| 180 |
ObjectRef::from_hex(&format!("{hh}{tail}")) |
| 181 |
} |
| 182 |
|
| 183 |
|
| 184 |
fn fanout_prefix(index: &str, r: &ObjectRef) -> String { |
| 185 |
let h = r.to_hex(); |
| 186 |
format!("{index}/{}/{}/", &h[..2], &h[2..]) |
| 187 |
} |
| 188 |
|
| 189 |
|
| 190 |
fn legacy_key(index: &str, r: &ObjectRef) -> String { |
| 191 |
let h = r.to_hex(); |
| 192 |
format!("{index}/{}/{}", &h[..2], &h[2..]) |
| 193 |
} |
| 194 |
|
| 195 |
fn now_ms() -> u64 { |
| 196 |
std::time::SystemTime::now() |
| 197 |
.duration_since(std::time::UNIX_EPOCH) |
| 198 |
.map(|d| d.as_millis() as u64) |
| 199 |
.unwrap_or(0) |
| 200 |
} |
| 201 |
|
| 202 |
#[derive(Clone, Copy, Debug)] |
| 203 |
struct PackEntry { |
| 204 |
offset: u64, |
| 205 |
stored_len: u32, |
| 206 |
raw_len: u32, |
| 207 |
ty: ObjectType, |
| 208 |
flags: u8, |
| 209 |
} |
| 210 |
|
| 211 |
|
| 212 |
struct PackIndex { |
| 213 |
|
| 214 |
key: String, |
| 215 |
refs: Vec<ObjectRef>, |
| 216 |
entries: Vec<PackEntry>, |
| 217 |
|
| 218 |
dict: Option<Arc<zstd::dict::DecoderDictionary<'static>>>, |
| 219 |
} |
| 220 |
|
| 221 |
impl std::fmt::Debug for PackIndex { |
| 222 |
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 223 |
f.debug_struct("PackIndex") |
| 224 |
.field("key", &self.key) |
| 225 |
.field("objects", &self.refs.len()) |
| 226 |
.field("dict", &self.dict.is_some()) |
| 227 |
.finish() |
| 228 |
} |
| 229 |
} |
| 230 |
|
| 231 |
|
| 232 |
fn decode_entry( |
| 233 |
r: &ObjectRef, |
| 234 |
e: PackEntry, |
| 235 |
stored: Vec<u8>, |
| 236 |
dict: Option<&Arc<zstd::dict::DecoderDictionary<'static>>>, |
| 237 |
) -> Result<Object> { |
| 238 |
let body = if e.flags & FLAG_DICT != 0 { |
| 239 |
let dict = dict.ok_or_else(|| StoreError::Corrupt(*r, NO_DICT.into()))?; |
| 240 |
zstd::bulk::Decompressor::with_prepared_dictionary(dict) |
| 241 |
.and_then(|mut d| d.decompress(&stored, e.raw_len as usize)) |
| 242 |
.map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))? |
| 243 |
} else if e.flags & FLAG_ZSTD != 0 { |
| 244 |
zstd::bulk::decompress(&stored, e.raw_len as usize) |
| 245 |
.map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))? |
| 246 |
} else { |
| 247 |
stored |
| 248 |
}; |
| 249 |
if body.len() != e.raw_len as usize || hash_object(e.ty, &body) != *r { |
| 250 |
return Err(StoreError::Corrupt( |
| 251 |
*r, |
| 252 |
"packed object does not match its reference".into(), |
| 253 |
)); |
| 254 |
} |
| 255 |
Ok((e.ty, body)) |
| 256 |
} |
| 257 |
|
| 258 |
impl PackIndex { |
| 259 |
|
| 260 |
|
| 261 |
|
| 262 |
fn load( |
| 263 |
backend: &Backend, |
| 264 |
idx_key: &str, |
| 265 |
pack_key: &str, |
| 266 |
pack_len: u64, |
| 267 |
) -> Result<Option<PackIndex>> { |
| 268 |
let data = match backend.get(idx_key) { |
| 269 |
Ok(d) => d, |
| 270 |
Err(BackendError::NotFound(_)) => return Ok(None), |
| 271 |
Err(e) => return Err(backend_err(e)), |
| 272 |
}; |
| 273 |
let bad = |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{idx_key}: {m}")); |
| 274 |
if data.len() < 12 || &data[..8] != IDX_MAGIC { |
| 275 |
return Err(bad("bad index header")); |
| 276 |
} |
| 277 |
let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize; |
| 278 |
if data.len() != 12 + count * IDX_ENTRY { |
| 279 |
return Err(bad("index length mismatch")); |
| 280 |
} |
| 281 |
let mut refs = Vec::with_capacity(count); |
| 282 |
let mut entries = Vec::with_capacity(count); |
| 283 |
for i in 0..count { |
| 284 |
let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY]; |
| 285 |
let r = ObjectRef::from_slice(&e[..32]).unwrap(); |
| 286 |
if let Some(prev) = refs.last() { |
| 287 |
if r <= *prev { |
| 288 |
return Err(bad("index not sorted")); |
| 289 |
} |
| 290 |
} |
| 291 |
refs.push(r); |
| 292 |
let entry = PackEntry { |
| 293 |
offset: u64::from_le_bytes(e[32..40].try_into().unwrap()), |
| 294 |
stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()), |
| 295 |
raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()), |
| 296 |
ty: ObjectType::ALL |
| 297 |
.get(e[48] as usize) |
| 298 |
.copied() |
| 299 |
.ok_or_else(|| bad("unknown object type"))?, |
| 300 |
flags: e[49], |
| 301 |
}; |
| 302 |
if entry.stored_len > MAX_OBJECT_LEN || entry.raw_len > MAX_OBJECT_LEN { |
| 303 |
return Err(StoreError::Corrupt( |
| 304 |
r, |
| 305 |
format!( |
| 306 |
"{idx_key}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be", |
| 307 |
entry.raw_len, entry.stored_len |
| 308 |
), |
| 309 |
)); |
| 310 |
} |
| 311 |
if entry.offset.saturating_add(entry.stored_len as u64) > pack_len { |
| 312 |
return Err(bad("index entry points outside the pack")); |
| 313 |
} |
| 314 |
entries.push(entry); |
| 315 |
} |
| 316 |
let dict_key = format!("{}.dict", pack_key.trim_end_matches(".pack")); |
| 317 |
let dict = get_optional(backend, &dict_key)? |
| 318 |
.map(|d| Arc::new(zstd::dict::DecoderDictionary::copy(&d))); |
| 319 |
Ok(Some(PackIndex { |
| 320 |
key: pack_key.to_string(), |
| 321 |
refs, |
| 322 |
entries, |
| 323 |
dict, |
| 324 |
})) |
| 325 |
} |
| 326 |
|
| 327 |
fn find(&self, r: &ObjectRef) -> Option<PackEntry> { |
| 328 |
self.refs.binary_search(r).ok().map(|i| self.entries[i]) |
| 329 |
} |
| 330 |
|
| 331 |
|
| 332 |
fn fetch(&self, backend: &Backend, r: &ObjectRef, e: PackEntry) -> Result<Object> { |
| 333 |
let stored = backend |
| 334 |
.get_range(&self.key, e.offset..e.offset + e.stored_len as u64) |
| 335 |
.map_err(|err| match err { |
| 336 |
BackendError::NotFound(_) => StoreError::NotFound(*r), |
| 337 |
other => backend_err(other), |
| 338 |
})?; |
| 339 |
decode_entry(r, e, stored, self.dict.as_ref()) |
| 340 |
} |
| 341 |
} |
| 342 |
|
| 343 |
fn type_index(ty: ObjectType) -> u8 { |
| 344 |
ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8 |
| 345 |
} |
| 346 |
|
| 347 |
|
| 348 |
#[derive(Debug, Clone, Default, PartialEq, Eq)] |
| 349 |
pub struct PackReport { |
| 350 |
pub objects: usize, |
| 351 |
pub raw_bytes: u64, |
| 352 |
pub packed_bytes: u64, |
| 353 |
} |
| 354 |
|
| 355 |
|
| 356 |
pub const MAX_PACKS: usize = 16; |
| 357 |
|
| 358 |
|
| 359 |
#[derive(Debug, Clone, Default, PartialEq, Eq)] |
| 360 |
pub struct PruneReport { |
| 361 |
pub objects: usize, |
| 362 |
pub bytes: u64, |
| 363 |
|
| 364 |
pub deferred: usize, |
| 365 |
} |
| 366 |
|
| 367 |
#[derive(Debug, thiserror::Error)] |
| 368 |
pub enum StoreError { |
| 369 |
#[error("io: {0}")] |
| 370 |
Io(#[from] io::Error), |
| 371 |
#[error("corrupt object {0}: {1}")] |
| 372 |
Corrupt(ObjectRef, String), |
| 373 |
#[error("object {0}")] |
| 374 |
Object(#[from] ObjectError), |
| 375 |
#[error("object {0} is a {1}, expected {2}")] |
| 376 |
WrongType(ObjectRef, &'static str, &'static str), |
| 377 |
#[error("object {0} not found")] |
| 378 |
NotFound(ObjectRef), |
| 379 |
#[error("a {kind} object of {size} bytes is larger than the {limit} bytes an object may be; nothing was stored. A change this large adds or touches too many files at once: commit it in parts (a directory at a time)")] |
| 380 |
TooLarge { |
| 381 |
kind: &'static str, |
| 382 |
size: usize, |
| 383 |
limit: u32, |
| 384 |
}, |
| 385 |
} |
| 386 |
|
| 387 |
pub type Result<T> = std::result::Result<T, StoreError>; |
| 388 |
|
| 389 |
pub struct Store { |
| 390 |
backend: Arc<Backend>, |
| 391 |
|
| 392 |
|
| 393 |
scratch: PathBuf, |
| 394 |
|
| 395 |
local: bool, |
| 396 |
packs: RefCell<Vec<Arc<PackIndex>>>, |
| 397 |
|
| 398 |
|
| 399 |
packs_listed: RefCell<Listed>, |
| 400 |
|
| 401 |
|
| 402 |
has_relisted: std::cell::Cell<Option<std::time::Instant>>, |
| 403 |
|
| 404 |
|
| 405 |
pending: RefCell<Option<Pending>>, |
| 406 |
batch_depth: std::cell::Cell<u32>, |
| 407 |
|
| 408 |
metas: RefCell<Option<crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta>>>, |
| 409 |
} |
| 410 |
|
| 411 |
impl std::fmt::Debug for Store { |
| 412 |
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 413 |
f.debug_struct("Store") |
| 414 |
.field("backend", &self.backend.describe()) |
| 415 |
.field("scratch", &self.scratch) |
| 416 |
.field("packs", &self.packs.borrow().len()) |
| 417 |
.finish() |
| 418 |
} |
| 419 |
} |
| 420 |
|
| 421 |
|
| 422 |
|
| 423 |
|
| 424 |
#[derive(Debug)] |
| 425 |
struct Pending { |
| 426 |
buffered: Vec<(ObjectType, Vec<u8>)>, |
| 427 |
buffered_bytes: usize, |
| 428 |
lookup: std::collections::HashMap<ObjectRef, PendingEntry>, |
| 429 |
pack: Option<PendingPack>, |
| 430 |
report: PackReport, |
| 431 |
} |
| 432 |
|
| 433 |
#[derive(Debug, Clone, Copy)] |
| 434 |
enum PendingEntry { |
| 435 |
Buffered(usize), |
| 436 |
Packed(PackEntry), |
| 437 |
} |
| 438 |
|
| 439 |
#[derive(Debug)] |
| 440 |
struct PendingPack { |
| 441 |
id: String, |
| 442 |
tmp_path: PathBuf, |
| 443 |
writer: io::BufWriter<fs::File>, |
| 444 |
offset: u64, |
| 445 |
index: Vec<(ObjectRef, PackEntry)>, |
| 446 |
} |
| 447 |
|
| 448 |
const SPILL_OBJECTS: usize = 256; |
| 449 |
const SPILL_BYTES: usize = 8 << 20; |
| 450 |
|
| 451 |
|
| 452 |
pub struct Batch<'a> { |
| 453 |
store: &'a Store, |
| 454 |
done: bool, |
| 455 |
} |
| 456 |
|
| 457 |
impl Batch<'_> { |
| 458 |
pub fn finish(mut self) -> Result<()> { |
| 459 |
self.done = true; |
| 460 |
self.store.finish_batch() |
| 461 |
} |
| 462 |
} |
| 463 |
|
| 464 |
impl Drop for Batch<'_> { |
| 465 |
fn drop(&mut self) { |
| 466 |
if !self.done { |
| 467 |
let _ = self.store.finish_batch(); |
| 468 |
} |
| 469 |
} |
| 470 |
} |
| 471 |
|
| 472 |
|
| 473 |
fn write_pack_entry( |
| 474 |
pack: &mut PendingPack, |
| 475 |
r: &ObjectRef, |
| 476 |
ty: ObjectType, |
| 477 |
raw_len: usize, |
| 478 |
flags: u8, |
| 479 |
stored: &[u8], |
| 480 |
report: &mut PackReport, |
| 481 |
) -> Result<PackEntry> { |
| 482 |
let header_len = 1 + 1 + 4 + 4 + 32; |
| 483 |
pack.writer.write_all(&[type_index(ty), flags])?; |
| 484 |
pack.writer |
| 485 |
.write_all(&(stored.len() as u32).to_le_bytes())?; |
| 486 |
pack.writer.write_all(&(raw_len as u32).to_le_bytes())?; |
| 487 |
pack.writer.write_all(&r.0)?; |
| 488 |
pack.writer.write_all(stored)?; |
| 489 |
let entry = PackEntry { |
| 490 |
offset: pack.offset + header_len, |
| 491 |
stored_len: stored.len() as u32, |
| 492 |
raw_len: raw_len as u32, |
| 493 |
ty, |
| 494 |
flags, |
| 495 |
}; |
| 496 |
pack.offset += header_len + stored.len() as u64; |
| 497 |
pack.index.push((*r, entry)); |
| 498 |
report.objects += 1; |
| 499 |
report.raw_bytes += raw_len as u64; |
| 500 |
report.packed_bytes += stored.len() as u64; |
| 501 |
Ok(entry) |
| 502 |
} |
| 503 |
|
| 504 |
|
| 505 |
fn fresh_id() -> String { |
| 506 |
let mut nonce = [0u8; 8]; |
| 507 |
rand::rngs::OsRng.fill_bytes(&mut nonce); |
| 508 |
hex::encode(nonce) |
| 509 |
} |
| 510 |
|
| 511 |
|
| 512 |
|
| 513 |
|
| 514 |
|
| 515 |
fn memory_backend_for(dir: &Path) -> Option<Arc<Backend>> { |
| 516 |
use std::collections::HashMap; |
| 517 |
use std::sync::{Mutex, OnceLock}; |
| 518 |
if std::env::var_os("OXY_STORE_BACKEND").as_deref() != Some(std::ffi::OsStr::new("memory")) { |
| 519 |
return None; |
| 520 |
} |
| 521 |
static STORES: OnceLock<Mutex<HashMap<PathBuf, Arc<MemoryStore>>>> = OnceLock::new(); |
| 522 |
let key = dir.canonicalize().unwrap_or_else(|_| dir.to_path_buf()); |
| 523 |
let mut map = STORES |
| 524 |
.get_or_init(Default::default) |
| 525 |
.lock() |
| 526 |
.unwrap_or_else(|p| p.into_inner()); |
| 527 |
let store = map |
| 528 |
.entry(key) |
| 529 |
.or_insert_with(|| Arc::new(MemoryStore::new())) |
| 530 |
.clone(); |
| 531 |
Some(store) |
| 532 |
} |
| 533 |
|
| 534 |
#[cfg(test)] |
| 535 |
thread_local! { |
| 536 |
|
| 537 |
|
| 538 |
|
| 539 |
pub(crate) static TEST_BACKEND: RefCell<Option<Arc<Backend>>> = const { RefCell::new(None) }; |
| 540 |
} |
| 541 |
|
| 542 |
impl Store { |
| 543 |
|
| 544 |
pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> { |
| 545 |
let dir = dir.into(); |
| 546 |
fs::create_dir_all(dir.join("loose"))?; |
| 547 |
#[cfg(test)] |
| 548 |
if let Some(backend) = TEST_BACKEND.with(|t| t.borrow().clone()) { |
| 549 |
return Store::open_on(backend, dir); |
| 550 |
} |
| 551 |
if let Some(backend) = memory_backend_for(&dir) { |
| 552 |
return Store::open_on(backend, dir); |
| 553 |
} |
| 554 |
let backend: Arc<Backend> = |
| 555 |
Arc::new(LocalStore::open(&dir).map_err(|e| io::Error::other(e.to_string()))?); |
| 556 |
let mut store = Store::open_on(backend, dir)?; |
| 557 |
store.local = true; |
| 558 |
Ok(store) |
| 559 |
} |
| 560 |
|
| 561 |
|
| 562 |
|
| 563 |
pub fn open_on(backend: Arc<Backend>, scratch: impl Into<PathBuf>) -> io::Result<Store> { |
| 564 |
let scratch = scratch.into(); |
| 565 |
fs::create_dir_all(&scratch)?; |
| 566 |
let store = Store { |
| 567 |
backend, |
| 568 |
scratch, |
| 569 |
local: false, |
| 570 |
packs: RefCell::new(Vec::new()), |
| 571 |
packs_listed: RefCell::new(Vec::new()), |
| 572 |
has_relisted: std::cell::Cell::new(None), |
| 573 |
pending: RefCell::new(None), |
| 574 |
batch_depth: std::cell::Cell::new(0), |
| 575 |
metas: RefCell::new(None), |
| 576 |
}; |
| 577 |
store |
| 578 |
.reload_packs() |
| 579 |
.map_err(|e| io::Error::other(e.to_string()))?; |
| 580 |
Ok(store) |
| 581 |
} |
| 582 |
|
| 583 |
|
| 584 |
pub fn backend(&self) -> &Arc<Backend> { |
| 585 |
&self.backend |
| 586 |
} |
| 587 |
|
| 588 |
|
| 589 |
|
| 590 |
pub fn is_local(&self) -> bool { |
| 591 |
self.local |
| 592 |
} |
| 593 |
|
| 594 |
|
| 595 |
|
| 596 |
|
| 597 |
pub fn refresh(&self) -> Result<()> { |
| 598 |
self.reload_packs() |
| 599 |
} |
| 600 |
|
| 601 |
|
| 602 |
fn list_packs(&self) -> Result<(Vec<strata::Entry>, Listed)> { |
| 603 |
let listed = self.backend.list("pack-").map_err(backend_err)?; |
| 604 |
let sig = listed |
| 605 |
.iter() |
| 606 |
.map(|e| (e.key.clone(), e.size, e.modified_ms)) |
| 607 |
.collect(); |
| 608 |
Ok((listed, sig)) |
| 609 |
} |
| 610 |
|
| 611 |
|
| 612 |
|
| 613 |
|
| 614 |
|
| 615 |
fn reload_packs(&self) -> Result<()> { |
| 616 |
let (listed, sig) = self.list_packs()?; |
| 617 |
self.load_listed(listed, sig) |
| 618 |
} |
| 619 |
|
| 620 |
fn load_listed(&self, mut listed: Vec<strata::Entry>, mut sig: Listed) -> Result<()> { |
| 621 |
let had: std::collections::HashMap<String, Arc<PackIndex>> = self |
| 622 |
.packs |
| 623 |
.borrow() |
| 624 |
.iter() |
| 625 |
.map(|p| (p.key.clone(), p.clone())) |
| 626 |
.collect(); |
| 627 |
let mut loaded: std::collections::HashMap<String, Arc<PackIndex>> = Default::default(); |
| 628 |
for attempt in 0..2 { |
| 629 |
let retired = self.retired_in(&listed)?; |
| 630 |
let mut idx_keys: Vec<&str> = listed |
| 631 |
.iter() |
| 632 |
.map(|e| e.key.as_str()) |
| 633 |
.filter(|k| k.ends_with(".idx")) |
| 634 |
.collect(); |
| 635 |
idx_keys.sort(); |
| 636 |
let mut packs = Vec::new(); |
| 637 |
let mut vanished = false; |
| 638 |
for idx in idx_keys { |
| 639 |
let pack = format!("{}.pack", idx.trim_end_matches(".idx")); |
| 640 |
if retired.contains_key(&pack) { |
| 641 |
continue; |
| 642 |
} |
| 643 |
let Some(p) = listed.iter().find(|e| e.key == pack) else { |
| 644 |
continue; |
| 645 |
}; |
| 646 |
if let Some(kept) = had.get(&pack).or_else(|| loaded.get(&pack)) { |
| 647 |
packs.push(kept.clone()); |
| 648 |
continue; |
| 649 |
} |
| 650 |
match PackIndex::load(self.backend.as_ref(), idx, &pack, p.size)? { |
| 651 |
Some(index) => { |
| 652 |
let index = Arc::new(index); |
| 653 |
loaded.insert(pack, index.clone()); |
| 654 |
packs.push(index); |
| 655 |
} |
| 656 |
None => vanished = true, |
| 657 |
} |
| 658 |
} |
| 659 |
if vanished && attempt == 0 { |
| 660 |
(listed, sig) = self.list_packs()?; |
| 661 |
continue; |
| 662 |
} |
| 663 |
*self.packs.borrow_mut() = packs; |
| 664 |
*self.packs_listed.borrow_mut() = sig; |
| 665 |
return Ok(()); |
| 666 |
} |
| 667 |
unreachable!("the second round always returns") |
| 668 |
} |
| 669 |
|
| 670 |
|
| 671 |
|
| 672 |
|
| 673 |
|
| 674 |
fn reload_if_changed(&self) -> Result<bool> { |
| 675 |
let (listed, sig) = self.list_packs()?; |
| 676 |
if *self.packs_listed.borrow() == sig { |
| 677 |
return Ok(false); |
| 678 |
} |
| 679 |
self.load_listed(listed, sig)?; |
| 680 |
Ok(true) |
| 681 |
} |
| 682 |
|
| 683 |
|
| 684 |
|
| 685 |
fn retired_in( |
| 686 |
&self, |
| 687 |
listed: &[strata::Entry], |
| 688 |
) -> Result<std::collections::BTreeMap<String, u64>> { |
| 689 |
match listed.iter().find(|e| e.key == RETIRED_KEY) { |
| 690 |
Some(e) if e.size > 0 => Ok(self.read_retired()?.0), |
| 691 |
_ => Ok(Default::default()), |
| 692 |
} |
| 693 |
} |
| 694 |
|
| 695 |
|
| 696 |
fn read_retired( |
| 697 |
&self, |
| 698 |
) -> Result<( |
| 699 |
std::collections::BTreeMap<String, u64>, |
| 700 |
Option<strata::Version>, |
| 701 |
)> { |
| 702 |
match self.backend.read_ref(RETIRED_KEY).map_err(backend_err)? { |
| 703 |
Some((bytes, v)) => Ok((parse_retired(&bytes), Some(v))), |
| 704 |
None => Ok((Default::default(), None)), |
| 705 |
} |
| 706 |
} |
| 707 |
|
| 708 |
|
| 709 |
|
| 710 |
|
| 711 |
fn update_retired( |
| 712 |
&self, |
| 713 |
mut edit: impl FnMut(&mut std::collections::BTreeMap<String, u64>), |
| 714 |
) -> Result<()> { |
| 715 |
for _ in 0..RETIRED_TRIES { |
| 716 |
let (mut m, version) = self.read_retired()?; |
| 717 |
let before = m.clone(); |
| 718 |
edit(&mut m); |
| 719 |
if m == before && version.is_some() { |
| 720 |
return Ok(()); |
| 721 |
} |
| 722 |
match self |
| 723 |
.backend |
| 724 |
.write_ref(RETIRED_KEY, &encode_retired(&m), version.as_ref()) |
| 725 |
{ |
| 726 |
Ok(_) => return Ok(()), |
| 727 |
Err(BackendError::Conflict(_)) => continue, |
| 728 |
Err(e) => return Err(backend_err(e)), |
| 729 |
} |
| 730 |
} |
| 731 |
Err(StoreError::Io(io::Error::other( |
| 732 |
"other writers kept changing the record of retired packs; run the command again", |
| 733 |
))) |
| 734 |
} |
| 735 |
|
| 736 |
|
| 737 |
|
| 738 |
|
| 739 |
|
| 740 |
fn retire(&self, keys: &[String]) -> Result<()> { |
| 741 |
if keys.is_empty() { |
| 742 |
return Ok(()); |
| 743 |
} |
| 744 |
let now = now_ms(); |
| 745 |
self.update_retired(|m| { |
| 746 |
for k in keys { |
| 747 |
m.entry(k.clone()).or_insert(now); |
| 748 |
} |
| 749 |
}) |
| 750 |
} |
| 751 |
|
| 752 |
|
| 753 |
|
| 754 |
|
| 755 |
|
| 756 |
|
| 757 |
pub fn expire_retired( |
| 758 |
&self, |
| 759 |
older_than: std::time::Duration, |
| 760 |
progress: Progress, |
| 761 |
) -> Result<RetiredReport> { |
| 762 |
let (listed, _) = self.list_packs()?; |
| 763 |
let retired = self.retired_in(&listed)?; |
| 764 |
let mut report = RetiredReport::default(); |
| 765 |
if retired.is_empty() { |
| 766 |
return Ok(report); |
| 767 |
} |
| 768 |
let now = now_ms(); |
| 769 |
let older_than = older_than.as_millis() as u64; |
| 770 |
let due: Vec<(String, u64)> = retired |
| 771 |
.iter() |
| 772 |
.filter(|(_, t)| now.saturating_sub(**t) >= older_than) |
| 773 |
.map(|(k, t)| (k.clone(), *t)) |
| 774 |
.collect(); |
| 775 |
report.kept = retired.len() - due.len(); |
| 776 |
if due.is_empty() { |
| 777 |
return Ok(report); |
| 778 |
} |
| 779 |
progress(&format!("deleting {} retired files", due.len())); |
| 780 |
|
| 781 |
self.reload_packs()?; |
| 782 |
for (key, _) in &due { |
| 783 |
if key.ends_with(".pack") { |
| 784 |
self.remove_pack(key)?; |
| 785 |
report.packs += 1; |
| 786 |
} else if let Some(r) = loose_ref(key) { |
| 787 |
if self.find_packed(&r).is_some() { |
| 788 |
delete(self.backend.as_ref(), key)?; |
| 789 |
report.loose += 1; |
| 790 |
} |
| 791 |
} |
| 792 |
} |
| 793 |
self.update_retired(|m| { |
| 794 |
for (k, t) in &due { |
| 795 |
if m.get(k) == Some(t) { |
| 796 |
m.remove(k); |
| 797 |
} |
| 798 |
} |
| 799 |
})?; |
| 800 |
self.reload_packs()?; |
| 801 |
Ok(report) |
| 802 |
} |
| 803 |
|
| 804 |
|
| 805 |
fn retired_loose(&self) -> Result<std::collections::HashSet<String>> { |
| 806 |
let (listed, _) = self.list_packs()?; |
| 807 |
Ok(self |
| 808 |
.retired_in(&listed)? |
| 809 |
.into_keys() |
| 810 |
.filter(|k| k.starts_with("loose/")) |
| 811 |
.collect()) |
| 812 |
} |
| 813 |
|
| 814 |
pub fn pack_count(&self) -> usize { |
| 815 |
self.packs.borrow().len() |
| 816 |
} |
| 817 |
|
| 818 |
fn find_packed(&self, r: &ObjectRef) -> Option<(usize, PackEntry)> { |
| 819 |
self.packs |
| 820 |
.borrow() |
| 821 |
.iter() |
| 822 |
.enumerate() |
| 823 |
.find_map(|(i, p)| p.find(r).map(|e| (i, e))) |
| 824 |
} |
| 825 |
|
| 826 |
|
| 827 |
|
| 828 |
fn loose_objects(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> { |
| 829 |
let retired = self.retired_loose()?; |
| 830 |
let mut out = self.loose_objects_all()?; |
| 831 |
out.retain(|(_, e)| !retired.contains(&e.key)); |
| 832 |
Ok(out) |
| 833 |
} |
| 834 |
|
| 835 |
|
| 836 |
fn loose_objects_all(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> { |
| 837 |
let mut out: Vec<(ObjectRef, strata::Entry)> = self |
| 838 |
.backend |
| 839 |
.list("loose/") |
| 840 |
.map_err(backend_err)? |
| 841 |
.into_iter() |
| 842 |
.filter_map(|e| loose_ref(&e.key).map(|r| (r, e))) |
| 843 |
.collect(); |
| 844 |
out.sort_by_key(|(r, _)| *r); |
| 845 |
Ok(out) |
| 846 |
} |
| 847 |
|
| 848 |
pub fn loose_count(&self) -> io::Result<usize> { |
| 849 |
self.loose_objects() |
| 850 |
.map(|l| l.len()) |
| 851 |
.map_err(|e| io::Error::other(e.to_string())) |
| 852 |
} |
| 853 |
|
| 854 |
|
| 855 |
|
| 856 |
pub fn pack_loose(&self) -> Result<PackReport> { |
| 857 |
let loose = self.loose_objects()?; |
| 858 |
self.write_pack( |
| 859 |
&loose.iter().map(|(r, _)| *r).collect::<Vec<_>>(), |
| 860 |
&loose.iter().map(|(_, e)| e.key.clone()).collect::<Vec<_>>(), |
| 861 |
&[], |
| 862 |
ZSTD_LEVEL, |
| 863 |
&mut silent(), |
| 864 |
) |
| 865 |
} |
| 866 |
|
| 867 |
|
| 868 |
|
| 869 |
|
| 870 |
pub fn repack_all(&self, level: i32) -> Result<(PackReport, usize)> { |
| 871 |
self.repack_all_with(level, &mut silent()) |
| 872 |
} |
| 873 |
|
| 874 |
|
| 875 |
pub fn repack_all_with(&self, level: i32, progress: Progress) -> Result<(PackReport, usize)> { |
| 876 |
let loose = self.loose_objects()?; |
| 877 |
let (mut refs, keys, trained): (Vec<ObjectRef>, Vec<String>, bool) = { |
| 878 |
let packs = self.packs.borrow(); |
| 879 |
( |
| 880 |
packs.iter().flat_map(|p| p.refs.iter().copied()).collect(), |
| 881 |
packs.iter().map(|p| p.key.clone()).collect(), |
| 882 |
packs.first().is_some_and(|p| p.dict.is_some()), |
| 883 |
) |
| 884 |
}; |
| 885 |
|
| 886 |
|
| 887 |
|
| 888 |
|
| 889 |
if loose.is_empty() && keys.len() <= 1 && (level == ZSTD_LEVEL || trained) { |
| 890 |
return Ok((PackReport::default(), 0)); |
| 891 |
} |
| 892 |
refs.extend(loose.iter().map(|(r, _)| *r)); |
| 893 |
let loose_keys: Vec<String> = loose.into_iter().map(|(_, e)| e.key).collect(); |
| 894 |
let n = keys.len(); |
| 895 |
let report = self.write_pack(&refs, &loose_keys, &keys, level, progress)?; |
| 896 |
Ok((report, n)) |
| 897 |
} |
| 898 |
|
| 899 |
|
| 900 |
|
| 901 |
pub fn merge_packs(&self) -> Result<usize> { |
| 902 |
let (refs, keys): (Vec<ObjectRef>, Vec<String>) = { |
| 903 |
let packs = self.packs.borrow(); |
| 904 |
if packs.len() <= MAX_PACKS { |
| 905 |
return Ok(0); |
| 906 |
} |
| 907 |
( |
| 908 |
packs.iter().flat_map(|p| p.refs.iter().copied()).collect(), |
| 909 |
packs.iter().map(|p| p.key.clone()).collect(), |
| 910 |
) |
| 911 |
}; |
| 912 |
let n = keys.len(); |
| 913 |
self.write_pack(&refs, &[], &keys, ZSTD_LEVEL, &mut silent())?; |
| 914 |
Ok(n) |
| 915 |
} |
| 916 |
|
| 917 |
|
| 918 |
|
| 919 |
|
| 920 |
|
| 921 |
|
| 922 |
pub fn prune( |
| 923 |
&self, |
| 924 |
keep: &std::collections::HashSet<ObjectRef>, |
| 925 |
grace: std::time::Duration, |
| 926 |
) -> Result<PruneReport> { |
| 927 |
self.prune_with(keep, grace, &mut silent()) |
| 928 |
} |
| 929 |
|
| 930 |
|
| 931 |
pub fn prune_with( |
| 932 |
&self, |
| 933 |
keep: &std::collections::HashSet<ObjectRef>, |
| 934 |
grace: std::time::Duration, |
| 935 |
progress: Progress, |
| 936 |
) -> Result<PruneReport> { |
| 937 |
progress("pruning loose objects"); |
| 938 |
let now = now_ms(); |
| 939 |
let grace_ms = grace.as_millis() as u64; |
| 940 |
|
| 941 |
let old_enough = |
| 942 |
|modified: Option<u64>| modified.is_none_or(|t| now.saturating_sub(t) >= grace_ms); |
| 943 |
let mut report = PruneReport::default(); |
| 944 |
|
| 945 |
|
| 946 |
let retired = self.retired_loose()?; |
| 947 |
let mut unretired: Vec<String> = Vec::new(); |
| 948 |
for (r, entry) in self.loose_objects_all()? { |
| 949 |
if keep.contains(&r) { |
| 950 |
continue; |
| 951 |
} |
| 952 |
if retired.contains(&entry.key) { |
| 953 |
if old_enough(entry.modified_ms) { |
| 954 |
delete(self.backend.as_ref(), &entry.key)?; |
| 955 |
unretired.push(entry.key); |
| 956 |
} |
| 957 |
continue; |
| 958 |
} |
| 959 |
if !old_enough(entry.modified_ms) { |
| 960 |
report.deferred += 1; |
| 961 |
continue; |
| 962 |
} |
| 963 |
report.bytes += entry.size; |
| 964 |
report.objects += 1; |
| 965 |
delete(self.backend.as_ref(), &entry.key)?; |
| 966 |
self.remove_sidecars(&r)?; |
| 967 |
} |
| 968 |
if !unretired.is_empty() { |
| 969 |
self.update_retired(|m| { |
| 970 |
for k in &unretired { |
| 971 |
m.remove(k); |
| 972 |
} |
| 973 |
})?; |
| 974 |
} |
| 975 |
|
| 976 |
let mut keep_refs: Vec<ObjectRef> = Vec::new(); |
| 977 |
let mut old_keys: Vec<String> = Vec::new(); |
| 978 |
let mut rewrite = false; |
| 979 |
progress("pruning packs"); |
| 980 |
for p in self.packs.borrow().iter() { |
| 981 |
let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count(); |
| 982 |
if unreachable == 0 { |
| 983 |
continue; |
| 984 |
} |
| 985 |
let modified = self |
| 986 |
.backend |
| 987 |
.stat(&p.key) |
| 988 |
.map_err(backend_err)? |
| 989 |
.and_then(|e| e.modified_ms); |
| 990 |
if !old_enough(modified) { |
| 991 |
report.deferred += unreachable; |
| 992 |
continue; |
| 993 |
} |
| 994 |
rewrite = true; |
| 995 |
old_keys.push(p.key.clone()); |
| 996 |
for (r, e) in p.refs.iter().zip(p.entries.iter()) { |
| 997 |
if keep.contains(r) { |
| 998 |
keep_refs.push(*r); |
| 999 |
} else { |
| 1000 |
report.objects += 1; |
| 1001 |
report.bytes += e.stored_len as u64; |
| 1002 |
} |
| 1003 |
} |
| 1004 |
} |
| 1005 |
if rewrite { |
| 1006 |
let dropped: Vec<ObjectRef> = { |
| 1007 |
let packs = self.packs.borrow(); |
| 1008 |
packs |
| 1009 |
.iter() |
| 1010 |
.filter(|p| old_keys.contains(&p.key)) |
| 1011 |
.flat_map(|p| p.refs.iter().copied()) |
| 1012 |
.filter(|r| !keep.contains(r)) |
| 1013 |
.collect() |
| 1014 |
}; |
| 1015 |
self.write_pack(&keep_refs, &[], &old_keys, ZSTD_LEVEL, progress)?; |
| 1016 |
for r in dropped { |
| 1017 |
self.remove_sidecars(&r)?; |
| 1018 |
} |
| 1019 |
} |
| 1020 |
Ok(report) |
| 1021 |
} |
| 1022 |
|
| 1023 |
|
| 1024 |
|
| 1025 |
fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> { |
| 1026 |
for index in ["sigs", "seals", "receipts"] { |
| 1027 |
delete(self.backend.as_ref(), &legacy_key(index, r))?; |
| 1028 |
} |
| 1029 |
for index in ["sig", "seal", "receipt"] { |
| 1030 |
for e in self |
| 1031 |
.backend |
| 1032 |
.list(&fanout_prefix(index, r)) |
| 1033 |
.map_err(backend_err)? |
| 1034 |
{ |
| 1035 |
delete(self.backend.as_ref(), &e.key)?; |
| 1036 |
} |
| 1037 |
} |
| 1038 |
Ok(()) |
| 1039 |
} |
| 1040 |
|
| 1041 |
|
| 1042 |
|
| 1043 |
|
| 1044 |
fn finish_pack_files( |
| 1045 |
&self, |
| 1046 |
id: &str, |
| 1047 |
tmp_pack: &Path, |
| 1048 |
index: &[(ObjectRef, PackEntry)], |
| 1049 |
) -> Result<()> { |
| 1050 |
let mut idx = Vec::with_capacity(12 + index.len() * IDX_ENTRY); |
| 1051 |
idx.extend_from_slice(IDX_MAGIC); |
| 1052 |
idx.extend_from_slice(&(index.len() as u32).to_le_bytes()); |
| 1053 |
for (r, e) in index { |
| 1054 |
idx.extend_from_slice(&r.0); |
| 1055 |
idx.extend_from_slice(&e.offset.to_le_bytes()); |
| 1056 |
idx.extend_from_slice(&e.stored_len.to_le_bytes()); |
| 1057 |
idx.extend_from_slice(&e.raw_len.to_le_bytes()); |
| 1058 |
idx.extend_from_slice(&[type_index(e.ty), e.flags]); |
| 1059 |
} |
| 1060 |
let put = self |
| 1061 |
.backend |
| 1062 |
.put_new_file(&format!("pack-{id}.pack"), tmp_pack) |
| 1063 |
.map_err(backend_err); |
| 1064 |
|
| 1065 |
|
| 1066 |
let _ = fs::remove_file(tmp_pack); |
| 1067 |
put?; |
| 1068 |
self.backend |
| 1069 |
.put_new(&format!("pack-{id}.idx"), &idx) |
| 1070 |
.map_err(backend_err)?; |
| 1071 |
self.reload_packs() |
| 1072 |
} |
| 1073 |
|
| 1074 |
|
| 1075 |
|
| 1076 |
fn remove_pack(&self, pack_key: &str) -> Result<()> { |
| 1077 |
let stem = pack_key.trim_end_matches(".pack"); |
| 1078 |
delete(self.backend.as_ref(), &format!("{stem}.idx"))?; |
| 1079 |
delete(self.backend.as_ref(), pack_key)?; |
| 1080 |
delete(self.backend.as_ref(), &format!("{stem}.dict")) |
| 1081 |
} |
| 1082 |
|
| 1083 |
|
| 1084 |
|
| 1085 |
|
| 1086 |
|
| 1087 |
fn write_pack( |
| 1088 |
&self, |
| 1089 |
refs: &[ObjectRef], |
| 1090 |
loose_keys: &[String], |
| 1091 |
old_packs: &[String], |
| 1092 |
level: i32, |
| 1093 |
progress: Progress, |
| 1094 |
) -> Result<PackReport> { |
| 1095 |
if refs.is_empty() { |
| 1096 |
self.retire(old_packs)?; |
| 1097 |
self.reload_packs()?; |
| 1098 |
return Ok(PackReport::default()); |
| 1099 |
} |
| 1100 |
let mut refs: Vec<ObjectRef> = refs.to_vec(); |
| 1101 |
refs.sort(); |
| 1102 |
refs.dedup(); |
| 1103 |
let id = fresh_id(); |
| 1104 |
let tmp_pack = self.scratch.join(format!(".tmp-{id}.pack")); |
| 1105 |
let mut report = PackReport::default(); |
| 1106 |
let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len()); |
| 1107 |
|
| 1108 |
let dict = if level == REPACK_ZSTD_LEVEL && refs.len() >= DICT_MIN_OBJECTS { |
| 1109 |
let _t = crate::timing::phase("pack: train a dictionary"); |
| 1110 |
progress(&format!("training a dictionary on {} objects", refs.len())); |
| 1111 |
self.train_dictionary(&refs) |
| 1112 |
} else { |
| 1113 |
None |
| 1114 |
}; |
| 1115 |
let encoder = dict |
| 1116 |
.as_ref() |
| 1117 |
.map(|d| zstd::dict::EncoderDictionary::copy(d, level)); |
| 1118 |
let nocompress = self.nocompress_set()?; |
| 1119 |
{ |
| 1120 |
let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?); |
| 1121 |
f.write_all(PACK_MAGIC)?; |
| 1122 |
let mut offset = PACK_MAGIC.len() as u64; |
| 1123 |
|
| 1124 |
|
| 1125 |
let mut batches = refs.chunks(1024); |
| 1126 |
let mut compressed_batch = Vec::new(); |
| 1127 |
let total = refs.len(); |
| 1128 |
let mut reported = usize::MAX; |
| 1129 |
loop { |
| 1130 |
if compressed_batch.is_empty() { |
| 1131 |
let Some(batch) = batches.next() else { break }; |
| 1132 |
|
| 1133 |
let tenth = report.objects * 10 / total.max(1); |
| 1134 |
if tenth != reported { |
| 1135 |
reported = tenth; |
| 1136 |
progress(&format!("packing {} of {total} objects", report.objects)); |
| 1137 |
} |
| 1138 |
let mut read: Vec<(ObjectRef, ObjectType, Vec<u8>, bool)> = |
| 1139 |
Vec::with_capacity(batch.len()); |
| 1140 |
for r in batch { |
| 1141 |
let (ty, body) = self.get(r)?; |
| 1142 |
read.push((*r, ty, body, nocompress.contains(r))); |
| 1143 |
} |
| 1144 |
use rayon::prelude::*; |
| 1145 |
compressed_batch = read |
| 1146 |
.into_par_iter() |
| 1147 |
.map_init( |
| 1148 |
|| { |
| 1149 |
encoder.as_ref().and_then(|e| { |
| 1150 |
zstd::bulk::Compressor::with_prepared_dictionary(e).ok() |
| 1151 |
}) |
| 1152 |
}, |
| 1153 |
|with_dict, (r, ty, body, raw)| { |
| 1154 |
let packed = if raw { |
| 1155 |
None |
| 1156 |
} else { |
| 1157 |
|
| 1158 |
|
| 1159 |
let dicted = with_dict |
| 1160 |
.as_mut() |
| 1161 |
.filter(|_| body.len() <= DICT_OBJECT_MAX) |
| 1162 |
.and_then(|c| c.compress(&body).ok()) |
| 1163 |
.map(|c| (FLAG_ZSTD | FLAG_DICT, c)); |
| 1164 |
dicted |
| 1165 |
.or_else(|| { |
| 1166 |
zstd::bulk::compress(&body, level) |
| 1167 |
.ok() |
| 1168 |
.map(|c| (FLAG_ZSTD, c)) |
| 1169 |
}) |
| 1170 |
.filter(|(_, c)| c.len() < body.len()) |
| 1171 |
}; |
| 1172 |
(r, ty, body, packed) |
| 1173 |
}, |
| 1174 |
) |
| 1175 |
.collect::<Vec<_>>(); |
| 1176 |
compressed_batch.reverse(); |
| 1177 |
} |
| 1178 |
let Some((r, ty, body, packed)) = compressed_batch.pop() else { |
| 1179 |
continue; |
| 1180 |
}; |
| 1181 |
let r = &r; |
| 1182 |
let (flags, stored) = match packed { |
| 1183 |
Some(c) => c, |
| 1184 |
None => (0, body.clone()), |
| 1185 |
}; |
| 1186 |
let header_len = 1 + 1 + 4 + 4 + 32; |
| 1187 |
f.write_all(&[type_index(ty), flags])?; |
| 1188 |
f.write_all(&(stored.len() as u32).to_le_bytes())?; |
| 1189 |
f.write_all(&(body.len() as u32).to_le_bytes())?; |
| 1190 |
f.write_all(&r.0)?; |
| 1191 |
f.write_all(&stored)?; |
| 1192 |
index.push(( |
| 1193 |
*r, |
| 1194 |
PackEntry { |
| 1195 |
offset: offset + header_len, |
| 1196 |
stored_len: stored.len() as u32, |
| 1197 |
raw_len: body.len() as u32, |
| 1198 |
ty, |
| 1199 |
flags, |
| 1200 |
}, |
| 1201 |
)); |
| 1202 |
offset += header_len + stored.len() as u64; |
| 1203 |
report.objects += 1; |
| 1204 |
report.raw_bytes += body.len() as u64; |
| 1205 |
report.packed_bytes += stored.len() as u64; |
| 1206 |
} |
| 1207 |
f.flush()?; |
| 1208 |
f.get_ref().sync_all()?; |
| 1209 |
} |
| 1210 |
|
| 1211 |
if let Some(d) = &dict { |
| 1212 |
self.backend |
| 1213 |
.put_new(&format!("pack-{id}.dict"), d) |
| 1214 |
.map_err(backend_err)?; |
| 1215 |
} |
| 1216 |
progress(&format!("writing the index of {} objects", index.len())); |
| 1217 |
self.finish_pack_files(&id, &tmp_pack, &index)?; |
| 1218 |
let mut retired = old_packs.to_vec(); |
| 1219 |
retired.extend_from_slice(loose_keys); |
| 1220 |
self.retire(&retired)?; |
| 1221 |
self.reload_packs()?; |
| 1222 |
Ok(report) |
| 1223 |
} |
| 1224 |
|
| 1225 |
|
| 1226 |
|
| 1227 |
fn train_dictionary(&self, refs: &[ObjectRef]) -> Option<Vec<u8>> { |
| 1228 |
let step = (refs.len() / 10_000).max(1); |
| 1229 |
let mut samples: Vec<Vec<u8>> = Vec::new(); |
| 1230 |
let mut total = 0usize; |
| 1231 |
for r in refs.iter().step_by(step) { |
| 1232 |
if total > 32 << 20 { |
| 1233 |
break; |
| 1234 |
} |
| 1235 |
let Ok((ty, body)) = self.get(r) else { |
| 1236 |
continue; |
| 1237 |
}; |
| 1238 |
if matches!( |
| 1239 |
ty, |
| 1240 |
ObjectType::Chunk | ObjectType::Blob | ObjectType::Change |
| 1241 |
) && !body.is_empty() |
| 1242 |
&& body.len() < 1 << 20 |
| 1243 |
{ |
| 1244 |
total += body.len(); |
| 1245 |
samples.push(body); |
| 1246 |
} |
| 1247 |
} |
| 1248 |
zstd::dict::from_samples(&samples, DICT_BYTES).ok() |
| 1249 |
} |
| 1250 |
|
| 1251 |
|
| 1252 |
pub fn dir(&self) -> &Path { |
| 1253 |
&self.scratch |
| 1254 |
} |
| 1255 |
|
| 1256 |
|
| 1257 |
|
| 1258 |
#[cfg(test)] |
| 1259 |
pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf { |
| 1260 |
self.scratch.join(loose_key(r)) |
| 1261 |
} |
| 1262 |
|
| 1263 |
|
| 1264 |
|
| 1265 |
#[cfg(test)] |
| 1266 |
pub(crate) fn remove_loose_for_test(&self, r: &ObjectRef) -> Result<()> { |
| 1267 |
delete(self.backend.as_ref(), &loose_key(r)) |
| 1268 |
} |
| 1269 |
|
| 1270 |
|
| 1271 |
|
| 1272 |
|
| 1273 |
|
| 1274 |
pub fn has(&self, r: &ObjectRef) -> bool { |
| 1275 |
if self.has_now(r) { |
| 1276 |
return true; |
| 1277 |
} |
| 1278 |
let recent = self |
| 1279 |
.has_relisted |
| 1280 |
.get() |
| 1281 |
.is_some_and(|t| t.elapsed() < HAS_RELIST_AFTER); |
| 1282 |
if recent { |
| 1283 |
return false; |
| 1284 |
} |
| 1285 |
match self.reload_if_changed() { |
| 1286 |
Ok(true) => { |
| 1287 |
self.has_relisted.set(None); |
| 1288 |
self.has_now(r) |
| 1289 |
} |
| 1290 |
_ => { |
| 1291 |
self.has_relisted.set(Some(std::time::Instant::now())); |
| 1292 |
false |
| 1293 |
} |
| 1294 |
} |
| 1295 |
} |
| 1296 |
|
| 1297 |
|
| 1298 |
fn has_now(&self, r: &ObjectRef) -> bool { |
| 1299 |
self.find_packed(r).is_some() |
| 1300 |
|| self |
| 1301 |
.pending |
| 1302 |
.borrow() |
| 1303 |
.as_ref() |
| 1304 |
.is_some_and(|p| p.lookup.contains_key(r)) |
| 1305 |
|| self.backend.has(&loose_key(r)).unwrap_or(false) |
| 1306 |
} |
| 1307 |
|
| 1308 |
|
| 1309 |
|
| 1310 |
|
| 1311 |
|
| 1312 |
pub fn begin_batch(&self) -> Result<Batch<'_>> { |
| 1313 |
let depth = self.batch_depth.get(); |
| 1314 |
if depth == 0 { |
| 1315 |
*self.pending.borrow_mut() = Some(Pending { |
| 1316 |
buffered: vec![], |
| 1317 |
buffered_bytes: 0, |
| 1318 |
lookup: Default::default(), |
| 1319 |
pack: None, |
| 1320 |
report: PackReport::default(), |
| 1321 |
}); |
| 1322 |
} |
| 1323 |
self.batch_depth.set(depth + 1); |
| 1324 |
Ok(Batch { |
| 1325 |
store: self, |
| 1326 |
done: false, |
| 1327 |
}) |
| 1328 |
} |
| 1329 |
|
| 1330 |
fn finish_batch(&self) -> Result<()> { |
| 1331 |
let depth = self.batch_depth.get(); |
| 1332 |
if depth == 0 { |
| 1333 |
return Ok(()); |
| 1334 |
} |
| 1335 |
self.batch_depth.set(depth - 1); |
| 1336 |
if depth > 1 { |
| 1337 |
return Ok(()); |
| 1338 |
} |
| 1339 |
let Some(pending) = self.pending.borrow_mut().take() else { |
| 1340 |
return Ok(()); |
| 1341 |
}; |
| 1342 |
self.write_pending(pending) |
| 1343 |
} |
| 1344 |
|
| 1345 |
|
| 1346 |
|
| 1347 |
|
| 1348 |
pub fn flush(&self) -> Result<()> { |
| 1349 |
let Some(pending) = self.pending.borrow_mut().take() else { |
| 1350 |
return Ok(()); |
| 1351 |
}; |
| 1352 |
self.write_pending(pending)?; |
| 1353 |
*self.pending.borrow_mut() = Some(Pending { |
| 1354 |
buffered: vec![], |
| 1355 |
buffered_bytes: 0, |
| 1356 |
lookup: Default::default(), |
| 1357 |
pack: None, |
| 1358 |
report: PackReport::default(), |
| 1359 |
}); |
| 1360 |
Ok(()) |
| 1361 |
} |
| 1362 |
|
| 1363 |
fn write_pending(&self, pending: Pending) -> Result<()> { |
| 1364 |
match pending.pack { |
| 1365 |
None => { |
| 1366 |
let objects: Vec<(ObjectType, &[u8])> = pending |
| 1367 |
.buffered |
| 1368 |
.iter() |
| 1369 |
.map(|(ty, body)| (*ty, body.as_slice())) |
| 1370 |
.collect(); |
| 1371 |
self.put_loose_many(&objects) |
| 1372 |
} |
| 1373 |
Some(mut pack) => { |
| 1374 |
pack.writer.flush()?; |
| 1375 |
pack.writer.get_ref().sync_all()?; |
| 1376 |
drop(pack.writer); |
| 1377 |
pack.index.sort_by_key(|(r, _)| *r); |
| 1378 |
self.finish_pack_files(&pack.id, &pack.tmp_path, &pack.index)?; |
| 1379 |
Ok(()) |
| 1380 |
} |
| 1381 |
} |
| 1382 |
} |
| 1383 |
|
| 1384 |
|
| 1385 |
fn pending_get(&self, r: &ObjectRef) -> Option<Result<Object>> { |
| 1386 |
let mut guard = self.pending.borrow_mut(); |
| 1387 |
let pending = guard.as_mut()?; |
| 1388 |
let entry = *pending.lookup.get(r)?; |
| 1389 |
Some(match entry { |
| 1390 |
PendingEntry::Buffered(i) => { |
| 1391 |
let (ty, body) = &pending.buffered[i]; |
| 1392 |
Ok((*ty, body.clone())) |
| 1393 |
} |
| 1394 |
PendingEntry::Packed(e) => { |
| 1395 |
let Some(pack) = pending.pack.as_mut() else { |
| 1396 |
return Some(Err(StoreError::NotFound(*r))); |
| 1397 |
}; |
| 1398 |
|
| 1399 |
(|| { |
| 1400 |
pack.writer.flush()?; |
| 1401 |
use std::io::{Read, Seek, SeekFrom}; |
| 1402 |
let mut f = fs::File::open(&pack.tmp_path)?; |
| 1403 |
f.seek(SeekFrom::Start(e.offset))?; |
| 1404 |
let mut stored = vec![0u8; e.stored_len as usize]; |
| 1405 |
f.read_exact(&mut stored)?; |
| 1406 |
decode_entry(r, e, stored, None) |
| 1407 |
})() |
| 1408 |
} |
| 1409 |
}) |
| 1410 |
} |
| 1411 |
|
| 1412 |
|
| 1413 |
fn pending_pack_write( |
| 1414 |
&self, |
| 1415 |
pack: &mut PendingPack, |
| 1416 |
r: &ObjectRef, |
| 1417 |
ty: ObjectType, |
| 1418 |
body: &[u8], |
| 1419 |
report: &mut PackReport, |
| 1420 |
) -> Result<PackEntry> { |
| 1421 |
let (flags, stored) = if self.is_nocompress(r) { |
| 1422 |
(0, body.to_vec()) |
| 1423 |
} else { |
| 1424 |
let compressed = zstd::bulk::compress(body, ZSTD_LEVEL) |
| 1425 |
.map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?; |
| 1426 |
if compressed.len() < body.len() { |
| 1427 |
(FLAG_ZSTD, compressed) |
| 1428 |
} else { |
| 1429 |
(0, body.to_vec()) |
| 1430 |
} |
| 1431 |
}; |
| 1432 |
write_pack_entry(pack, r, ty, body.len(), flags, &stored, report) |
| 1433 |
} |
| 1434 |
|
| 1435 |
fn put_pending(&self, r: ObjectRef, ty: ObjectType, body: &[u8]) -> Result<()> { |
| 1436 |
let mut guard = self.pending.borrow_mut(); |
| 1437 |
let pending = guard.as_mut().expect("batch open"); |
| 1438 |
if pending.lookup.contains_key(&r) { |
| 1439 |
return Ok(()); |
| 1440 |
} |
| 1441 |
if pending.pack.is_none() |
| 1442 |
&& pending.buffered.len() < SPILL_OBJECTS |
| 1443 |
&& pending.buffered_bytes + body.len() <= SPILL_BYTES |
| 1444 |
{ |
| 1445 |
pending.buffered.push((ty, body.to_vec())); |
| 1446 |
pending.buffered_bytes += body.len(); |
| 1447 |
pending |
| 1448 |
.lookup |
| 1449 |
.insert(r, PendingEntry::Buffered(pending.buffered.len() - 1)); |
| 1450 |
return Ok(()); |
| 1451 |
} |
| 1452 |
self.ensure_pack(pending)?; |
| 1453 |
let mut report = std::mem::take(&mut pending.report); |
| 1454 |
let pack = pending.pack.as_mut().unwrap(); |
| 1455 |
let e = self.pending_pack_write(pack, &r, ty, body, &mut report)?; |
| 1456 |
pending.report = report; |
| 1457 |
pending.lookup.insert(r, PendingEntry::Packed(e)); |
| 1458 |
Ok(()) |
| 1459 |
} |
| 1460 |
|
| 1461 |
|
| 1462 |
|
| 1463 |
fn ensure_pack(&self, pending: &mut Pending) -> Result<()> { |
| 1464 |
if pending.pack.is_none() { |
| 1465 |
|
| 1466 |
let id = fresh_id(); |
| 1467 |
let tmp_path = self.scratch.join(format!(".tmp-{id}.pack")); |
| 1468 |
let mut writer = io::BufWriter::new(fs::File::create(&tmp_path)?); |
| 1469 |
writer.write_all(PACK_MAGIC)?; |
| 1470 |
let mut pack = PendingPack { |
| 1471 |
id, |
| 1472 |
tmp_path, |
| 1473 |
writer, |
| 1474 |
offset: PACK_MAGIC.len() as u64, |
| 1475 |
index: vec![], |
| 1476 |
}; |
| 1477 |
let buffered = std::mem::take(&mut pending.buffered); |
| 1478 |
pending.buffered_bytes = 0; |
| 1479 |
let mut report = std::mem::take(&mut pending.report); |
| 1480 |
for (bty, bbody) in &buffered { |
| 1481 |
let br = hash_object(*bty, bbody); |
| 1482 |
let e = self.pending_pack_write(&mut pack, &br, *bty, bbody, &mut report)?; |
| 1483 |
pending.lookup.insert(br, PendingEntry::Packed(e)); |
| 1484 |
} |
| 1485 |
pending.report = report; |
| 1486 |
pending.pack = Some(pack); |
| 1487 |
} |
| 1488 |
Ok(()) |
| 1489 |
} |
| 1490 |
|
| 1491 |
|
| 1492 |
|
| 1493 |
|
| 1494 |
pub fn put_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<Vec<ObjectRef>> { |
| 1495 |
use rayon::prelude::*; |
| 1496 |
let total: usize = objects.iter().map(|(_, b)| b.len()).sum(); |
| 1497 |
if self.pending.borrow().is_none() || total < SPILL_BYTES { |
| 1498 |
return objects.iter().map(|(ty, b)| self.put(*ty, b)).collect(); |
| 1499 |
} |
| 1500 |
for (ty, b) in objects { |
| 1501 |
if b.len() > MAX_OBJECT_LEN as usize { |
| 1502 |
return Err(StoreError::TooLarge { |
| 1503 |
kind: ty.name(), |
| 1504 |
size: b.len(), |
| 1505 |
limit: MAX_OBJECT_LEN, |
| 1506 |
}); |
| 1507 |
} |
| 1508 |
} |
| 1509 |
let refs: Vec<ObjectRef> = objects |
| 1510 |
.par_iter() |
| 1511 |
.map(|(ty, b)| hash_object(*ty, b)) |
| 1512 |
.collect(); |
| 1513 |
|
| 1514 |
let mut seen: std::collections::HashSet<ObjectRef> = std::collections::HashSet::new(); |
| 1515 |
let mut todo: Vec<usize> = Vec::new(); |
| 1516 |
for (i, r) in refs.iter().enumerate() { |
| 1517 |
let known = self |
| 1518 |
.pending |
| 1519 |
.borrow() |
| 1520 |
.as_ref() |
| 1521 |
.is_some_and(|p| p.lookup.contains_key(r)); |
| 1522 |
if seen.insert(*r) |
| 1523 |
&& !known |
| 1524 |
&& self.find_packed(r).is_none() |
| 1525 |
&& !self.backend.has(&loose_key(r)).unwrap_or(false) |
| 1526 |
{ |
| 1527 |
todo.push(i); |
| 1528 |
} |
| 1529 |
} |
| 1530 |
let raw = self.nocompress_set()?; |
| 1531 |
let packed: Vec<(u8, Vec<u8>)> = todo |
| 1532 |
.par_iter() |
| 1533 |
.map(|i| { |
| 1534 |
let body = objects[*i].1; |
| 1535 |
if raw.contains(&refs[*i]) { |
| 1536 |
return (0, body.to_vec()); |
| 1537 |
} |
| 1538 |
match zstd::bulk::compress(body, ZSTD_LEVEL) { |
| 1539 |
Ok(c) if c.len() < body.len() => (FLAG_ZSTD, c), |
| 1540 |
_ => (0, body.to_vec()), |
| 1541 |
} |
| 1542 |
}) |
| 1543 |
.collect(); |
| 1544 |
let mut guard = self.pending.borrow_mut(); |
| 1545 |
let pending = guard.as_mut().expect("batch open"); |
| 1546 |
self.ensure_pack(pending)?; |
| 1547 |
let mut report = std::mem::take(&mut pending.report); |
| 1548 |
let pack = pending.pack.as_mut().unwrap(); |
| 1549 |
for (i, (flags, stored)) in todo.iter().zip(packed) { |
| 1550 |
let (ty, body) = objects[*i]; |
| 1551 |
let e = write_pack_entry(pack, &refs[*i], ty, body.len(), flags, &stored, &mut report)?; |
| 1552 |
pending.lookup.insert(refs[*i], PendingEntry::Packed(e)); |
| 1553 |
} |
| 1554 |
pending.report = report; |
| 1555 |
Ok(refs) |
| 1556 |
} |
| 1557 |
|
| 1558 |
|
| 1559 |
pub fn put(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> { |
| 1560 |
|
| 1561 |
|
| 1562 |
|
| 1563 |
if body.len() > MAX_OBJECT_LEN as usize { |
| 1564 |
return Err(StoreError::TooLarge { |
| 1565 |
kind: ty.name(), |
| 1566 |
size: body.len(), |
| 1567 |
limit: MAX_OBJECT_LEN, |
| 1568 |
}); |
| 1569 |
} |
| 1570 |
let r = hash_object(ty, body); |
| 1571 |
if self.find_packed(&r).is_some() || self.backend.has(&loose_key(&r)).unwrap_or(false) { |
| 1572 |
return Ok(r); |
| 1573 |
} |
| 1574 |
if self.pending.borrow().is_some() { |
| 1575 |
self.put_pending(r, ty, body)?; |
| 1576 |
return Ok(r); |
| 1577 |
} |
| 1578 |
self.put_loose(ty, body)?; |
| 1579 |
Ok(r) |
| 1580 |
} |
| 1581 |
|
| 1582 |
|
| 1583 |
fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> { |
| 1584 |
let r = hash_object(ty, body); |
| 1585 |
let mut bytes = header(ty, body.len()); |
| 1586 |
bytes.extend_from_slice(body); |
| 1587 |
crate::durable::step("write loose object")?; |
| 1588 |
put_new_or_existing(self.backend.as_ref(), &loose_key(&r), &bytes)?; |
| 1589 |
Ok(r) |
| 1590 |
} |
| 1591 |
|
| 1592 |
|
| 1593 |
|
| 1594 |
|
| 1595 |
|
| 1596 |
fn put_loose_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<()> { |
| 1597 |
if objects.len() <= 1 { |
| 1598 |
for (ty, body) in objects { |
| 1599 |
self.put_loose(*ty, body)?; |
| 1600 |
} |
| 1601 |
return Ok(()); |
| 1602 |
} |
| 1603 |
let mut seen = std::collections::HashSet::new(); |
| 1604 |
let items: Vec<(String, Vec<u8>)> = objects |
| 1605 |
.iter() |
| 1606 |
.filter_map(|(ty, body)| { |
| 1607 |
let r = hash_object(*ty, body); |
| 1608 |
seen.insert(r).then(|| { |
| 1609 |
let mut bytes = header(*ty, body.len()); |
| 1610 |
bytes.extend_from_slice(body); |
| 1611 |
(loose_key(&r), bytes) |
| 1612 |
}) |
| 1613 |
}) |
| 1614 |
.collect(); |
| 1615 |
self.backend.put_many_new(&items).map_err(backend_err) |
| 1616 |
} |
| 1617 |
|
| 1618 |
|
| 1619 |
|
| 1620 |
|
| 1621 |
|
| 1622 |
|
| 1623 |
|
| 1624 |
|
| 1625 |
|
| 1626 |
|
| 1627 |
|
| 1628 |
pub fn get(&self, r: &ObjectRef) -> Result<Object> { |
| 1629 |
match self.get_now(r) { |
| 1630 |
Err(e) if is_miss(&e) => { |
| 1631 |
|
| 1632 |
let packed_miss = |
| 1633 |
!matches!(e, StoreError::NotFound(_)) || self.find_packed(r).is_some(); |
| 1634 |
let changed = if packed_miss { |
| 1635 |
self.reload_packs()?; |
| 1636 |
true |
| 1637 |
} else { |
| 1638 |
self.reload_if_changed()? |
| 1639 |
}; |
| 1640 |
if changed { |
| 1641 |
self.get_now(r) |
| 1642 |
} else { |
| 1643 |
Err(e) |
| 1644 |
} |
| 1645 |
} |
| 1646 |
other => other, |
| 1647 |
} |
| 1648 |
} |
| 1649 |
|
| 1650 |
|
| 1651 |
fn get_now(&self, r: &ObjectRef) -> Result<Object> { |
| 1652 |
if let Some((i, e)) = self.find_packed(r) { |
| 1653 |
let pack = self.packs.borrow()[i].clone(); |
| 1654 |
return pack.fetch(self.backend.as_ref(), r, e); |
| 1655 |
} |
| 1656 |
match get_optional(self.backend.as_ref(), &loose_key(r))? { |
| 1657 |
Some(bytes) => parse_loose(&bytes, r), |
| 1658 |
None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))), |
| 1659 |
} |
| 1660 |
} |
| 1661 |
|
| 1662 |
|
| 1663 |
|
| 1664 |
|
| 1665 |
|
| 1666 |
|
| 1667 |
pub fn get_many(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> { |
| 1668 |
let mut out = self.get_many_now(refs); |
| 1669 |
let missed: Vec<usize> = out |
| 1670 |
.iter() |
| 1671 |
.enumerate() |
| 1672 |
.filter(|(_, g)| g.as_ref().is_err_and(is_miss)) |
| 1673 |
.map(|(i, _)| i) |
| 1674 |
.collect(); |
| 1675 |
if missed.is_empty() { |
| 1676 |
return out; |
| 1677 |
} |
| 1678 |
let packed_miss = missed.iter().any(|i| { |
| 1679 |
!matches!(out[*i], Err(StoreError::NotFound(_))) |
| 1680 |
|| self.find_packed(&refs[*i]).is_some() |
| 1681 |
}); |
| 1682 |
let changed = if packed_miss { |
| 1683 |
self.reload_packs().map(|_| true) |
| 1684 |
} else { |
| 1685 |
self.reload_if_changed() |
| 1686 |
}; |
| 1687 |
match changed { |
| 1688 |
Ok(true) => { |
| 1689 |
let again: Vec<ObjectRef> = missed.iter().map(|i| refs[*i]).collect(); |
| 1690 |
for (i, g) in missed.into_iter().zip(self.get_many_now(&again)) { |
| 1691 |
out[i] = g; |
| 1692 |
} |
| 1693 |
} |
| 1694 |
Ok(false) => {} |
| 1695 |
Err(e) => { |
| 1696 |
let why = e.to_string(); |
| 1697 |
for i in missed { |
| 1698 |
out[i] = Err(StoreError::Io(io::Error::other(why.clone()))); |
| 1699 |
} |
| 1700 |
} |
| 1701 |
} |
| 1702 |
out |
| 1703 |
} |
| 1704 |
|
| 1705 |
|
| 1706 |
fn get_many_now(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> { |
| 1707 |
use rayon::prelude::*; |
| 1708 |
let backend = self.backend.as_ref(); |
| 1709 |
let packs = self.packs.borrow(); |
| 1710 |
let place: Vec<Option<(usize, PackEntry)>> = refs |
| 1711 |
.iter() |
| 1712 |
.map(|r| { |
| 1713 |
packs |
| 1714 |
.iter() |
| 1715 |
.enumerate() |
| 1716 |
.find_map(|(p, pack)| pack.find(r).map(|e| (p, e))) |
| 1717 |
}) |
| 1718 |
.collect(); |
| 1719 |
let indexes: &[Arc<PackIndex>] = &packs; |
| 1720 |
let found: Vec<Option<Result<Object>>> = refs |
| 1721 |
.par_iter() |
| 1722 |
.zip(place.par_iter()) |
| 1723 |
.map(|(r, at)| match at { |
| 1724 |
Some((p, e)) => Some(indexes[*p].fetch(backend, r, *e)), |
| 1725 |
None => match get_optional(backend, &loose_key(r)) { |
| 1726 |
Ok(Some(bytes)) => Some(parse_loose(&bytes, r)), |
| 1727 |
Ok(None) => None, |
| 1728 |
Err(e) => Some(Err(e)), |
| 1729 |
}, |
| 1730 |
}) |
| 1731 |
.collect(); |
| 1732 |
drop(packs); |
| 1733 |
found |
| 1734 |
.into_iter() |
| 1735 |
.zip(refs) |
| 1736 |
.map(|(f, r)| { |
| 1737 |
f.unwrap_or_else(|| self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r)))) |
| 1738 |
}) |
| 1739 |
.collect() |
| 1740 |
} |
| 1741 |
|
| 1742 |
pub fn get_typed(&self, r: &ObjectRef, want: ObjectType) -> Result<Vec<u8>> { |
| 1743 |
let (ty, body) = self.get(r)?; |
| 1744 |
if ty != want { |
| 1745 |
return Err(StoreError::WrongType(*r, ty.name(), want.name())); |
| 1746 |
} |
| 1747 |
Ok(body) |
| 1748 |
} |
| 1749 |
|
| 1750 |
|
| 1751 |
|
| 1752 |
pub fn put_change(&self, c: &Change) -> Result<ObjectRef> { |
| 1753 |
let encoding = crate::timing::phase("store: encode the change and its texts"); |
| 1754 |
let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?; |
| 1755 |
drop(encoding); |
| 1756 |
let _t = crate::timing::phase("store: put the encoding"); |
| 1757 |
self.put(ObjectType::Change, &bytes) |
| 1758 |
} |
| 1759 |
|
| 1760 |
|
| 1761 |
|
| 1762 |
pub fn put_texts(&self, texts: &[crate::objects::StoredText]) -> Result<()> { |
| 1763 |
let mut chunks: Vec<(ObjectType, &[u8])> = Vec::new(); |
| 1764 |
for t in texts { |
| 1765 |
if !self.has(&t.blob.hash()) { |
| 1766 |
chunks.extend( |
| 1767 |
crate::chunk::chunks(&t.content) |
| 1768 |
.into_iter() |
| 1769 |
.map(|c| (ObjectType::Chunk, c)), |
| 1770 |
); |
| 1771 |
} |
| 1772 |
} |
| 1773 |
self.put_many(&chunks)?; |
| 1774 |
for t in texts { |
| 1775 |
self.put(ObjectType::Blob, &t.blob.encode())?; |
| 1776 |
} |
| 1777 |
Ok(()) |
| 1778 |
} |
| 1779 |
|
| 1780 |
pub fn get_change(&self, r: &ObjectRef) -> Result<Change> { |
| 1781 |
Ok(self.get_change_parts(r)?.0) |
| 1782 |
} |
| 1783 |
|
| 1784 |
|
| 1785 |
|
| 1786 |
pub fn get_change_parts(&self, r: &ObjectRef) -> Result<(Change, Vec<ObjectRef>)> { |
| 1787 |
let bytes = self.get_typed(r, ObjectType::Change)?; |
| 1788 |
let mut parts = Vec::new(); |
| 1789 |
let mut failed = None; |
| 1790 |
let decoded = Change::decode_with(&bytes, &mut |text, count, form| match self |
| 1791 |
.read_text(text) |
| 1792 |
{ |
| 1793 |
Ok((blob, content)) => { |
| 1794 |
parts.push(*text); |
| 1795 |
parts.extend(blob.chunks); |
| 1796 |
crate::objects::text_lines(&content, count, form) |
| 1797 |
} |
| 1798 |
Err(e) => { |
| 1799 |
let why = e.to_string(); |
| 1800 |
failed = Some(e); |
| 1801 |
Err(ObjectError::Schema(format!( |
| 1802 |
"text blob {}: {why}", |
| 1803 |
text.short(12) |
| 1804 |
))) |
| 1805 |
} |
| 1806 |
}); |
| 1807 |
match (decoded, failed) { |
| 1808 |
(Ok(c), _) => Ok((c, parts)), |
| 1809 |
(Err(_), Some(StoreError::NotFound(t))) => Err(StoreError::Corrupt( |
| 1810 |
*r, |
| 1811 |
format!("its text blob {} is missing", t.short(12)), |
| 1812 |
)), |
| 1813 |
(Err(e), _) => Err(StoreError::Corrupt(*r, e.to_string())), |
| 1814 |
} |
| 1815 |
} |
| 1816 |
|
| 1817 |
|
| 1818 |
|
| 1819 |
|
| 1820 |
|
| 1821 |
pub fn get_change_lazy(&self, r: &ObjectRef) -> Result<(Change, Vec<crate::objects::TextRef>)> { |
| 1822 |
let bytes = self.get_typed(r, ObjectType::Change)?; |
| 1823 |
Change::decode_skeleton(&bytes, *r).map_err(|e| StoreError::Corrupt(*r, e.to_string())) |
| 1824 |
} |
| 1825 |
|
| 1826 |
|
| 1827 |
|
| 1828 |
|
| 1829 |
|
| 1830 |
|
| 1831 |
|
| 1832 |
|
| 1833 |
pub fn get_change_meta(&self, r: &ObjectRef) -> Result<crate::state::ChangeMeta> { |
| 1834 |
if self.metas.borrow().is_none() { |
| 1835 |
*self.metas.borrow_mut() = Some(self.read_metas_cache()); |
| 1836 |
} |
| 1837 |
let cached = self.metas.borrow().as_ref().and_then(|m| m.get(r).cloned()); |
| 1838 |
if let Some(meta) = cached { |
| 1839 |
if self.has(r) { |
| 1840 |
#[cfg(test)] |
| 1841 |
assert_eq!( |
| 1842 |
meta, |
| 1843 |
Change::decode_meta(&self.get_typed(r, ObjectType::Change)?).unwrap(), |
| 1844 |
"the metadata cache holds a change's metadata" |
| 1845 |
); |
| 1846 |
return Ok(meta); |
| 1847 |
} |
| 1848 |
} |
| 1849 |
let bytes = self.get_typed(r, ObjectType::Change)?; |
| 1850 |
let meta = |
| 1851 |
Change::decode_meta(&bytes).map_err(|e| StoreError::Corrupt(*r, e.to_string()))?; |
| 1852 |
self.remember_meta(r, &meta); |
| 1853 |
Ok(meta) |
| 1854 |
} |
| 1855 |
|
| 1856 |
|
| 1857 |
|
| 1858 |
|
| 1859 |
|
| 1860 |
|
| 1861 |
fn read_metas_cache(&self) -> crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta> { |
| 1862 |
let mut out = crate::fasthash::FastMap::default(); |
| 1863 |
let Ok(bytes) = fs::read(self.scratch.join("metas")) else { |
| 1864 |
return out; |
| 1865 |
}; |
| 1866 |
let mut r = crate::flat::Reader::new(&bytes); |
| 1867 |
let mut good = 0; |
| 1868 |
while !r.at_end() { |
| 1869 |
let mut record = || -> std::result::Result<_, ObjectError> { |
| 1870 |
let h = r.object()?; |
| 1871 |
let body = r.bytes()?; |
| 1872 |
let check = r.fixed::<8>()?; |
| 1873 |
Ok((h, body, check)) |
| 1874 |
}; |
| 1875 |
let Ok((h, body, check)) = record() else { |
| 1876 |
break; |
| 1877 |
}; |
| 1878 |
let mut sum = blake3::Hasher::new(); |
| 1879 |
sum.update(&h.0).update(body); |
| 1880 |
if sum.finalize().as_bytes()[..8] != check { |
| 1881 |
break; |
| 1882 |
} |
| 1883 |
match crate::state::ChangeMeta::read_flat(&mut crate::flat::Reader::new(body)) { |
| 1884 |
Ok(m) => out.insert(h, m), |
| 1885 |
Err(_) => break, |
| 1886 |
}; |
| 1887 |
good += 32 + 4 + body.len() + 8; |
| 1888 |
} |
| 1889 |
|
| 1890 |
if good < bytes.len() { |
| 1891 |
let _ = fs::OpenOptions::new() |
| 1892 |
.write(true) |
| 1893 |
.open(self.scratch.join("metas")) |
| 1894 |
.and_then(|f| f.set_len(good as u64)); |
| 1895 |
} |
| 1896 |
out |
| 1897 |
} |
| 1898 |
|
| 1899 |
|
| 1900 |
|
| 1901 |
fn remember_meta(&self, r: &ObjectRef, meta: &crate::state::ChangeMeta) { |
| 1902 |
let mut body = crate::flat::Writer::default(); |
| 1903 |
meta.write_flat(&mut body); |
| 1904 |
let mut w = crate::flat::Writer::default(); |
| 1905 |
w.object(r); |
| 1906 |
w.bytes(&body.0); |
| 1907 |
let mut sum = blake3::Hasher::new(); |
| 1908 |
sum.update(&r.0).update(&body.0); |
| 1909 |
w.fixed(&sum.finalize().as_bytes()[..8]); |
| 1910 |
let _ = fs::OpenOptions::new() |
| 1911 |
.create(true) |
| 1912 |
.append(true) |
| 1913 |
.open(self.scratch.join("metas")) |
| 1914 |
.and_then(|mut f| f.write_all(&w.0)); |
| 1915 |
if let Some(m) = self.metas.borrow_mut().as_mut() { |
| 1916 |
m.insert(*r, meta.clone()); |
| 1917 |
} |
| 1918 |
} |
| 1919 |
|
| 1920 |
fn read_text(&self, r: &ObjectRef) -> Result<(Blob, Vec<u8>)> { |
| 1921 |
let blob = Blob::decode(&self.get_typed(r, ObjectType::Blob)?)?; |
| 1922 |
let mut content = Vec::with_capacity(blob.length as usize); |
| 1923 |
for c in &blob.chunks { |
| 1924 |
content.extend(self.get_typed(c, ObjectType::Chunk)?); |
| 1925 |
} |
| 1926 |
Ok((blob, content)) |
| 1927 |
} |
| 1928 |
|
| 1929 |
pub fn put_chunk(&self, bytes: &[u8]) -> Result<ObjectRef> { |
| 1930 |
self.put(ObjectType::Chunk, bytes) |
| 1931 |
} |
| 1932 |
|
| 1933 |
|
| 1934 |
pub fn put_chunk_nocompress(&self, bytes: &[u8]) -> Result<ObjectRef> { |
| 1935 |
let r = hash_object(ObjectType::Chunk, bytes); |
| 1936 |
put_new_or_existing( |
| 1937 |
self.backend.as_ref(), |
| 1938 |
&format!("nocompress/{}", r.to_hex()), |
| 1939 |
b"", |
| 1940 |
)?; |
| 1941 |
self.put(ObjectType::Chunk, bytes) |
| 1942 |
} |
| 1943 |
|
| 1944 |
fn is_nocompress(&self, r: &ObjectRef) -> bool { |
| 1945 |
self.backend |
| 1946 |
.has(&format!("nocompress/{}", r.to_hex())) |
| 1947 |
.unwrap_or(false) |
| 1948 |
} |
| 1949 |
|
| 1950 |
|
| 1951 |
fn nocompress_set(&self) -> Result<std::collections::HashSet<ObjectRef>> { |
| 1952 |
Ok(self |
| 1953 |
.backend |
| 1954 |
.list("nocompress/") |
| 1955 |
.map_err(backend_err)? |
| 1956 |
.into_iter() |
| 1957 |
.filter_map(|e| ObjectRef::from_hex(e.key.trim_start_matches("nocompress/"))) |
| 1958 |
.collect()) |
| 1959 |
} |
| 1960 |
|
| 1961 |
pub fn put_state(&self, st: &StateObject) -> Result<ObjectRef> { |
| 1962 |
self.put(ObjectType::State, &st.encode()) |
| 1963 |
} |
| 1964 |
|
| 1965 |
pub fn get_state(&self, r: &ObjectRef) -> Result<StateObject> { |
| 1966 |
Ok(StateObject::decode(&self.get_typed(r, ObjectType::State)?)?) |
| 1967 |
} |
| 1968 |
|
| 1969 |
pub fn put_attest(&self, a: &Attest) -> Result<ObjectRef> { |
| 1970 |
self.put(ObjectType::Attest, &a.encode()) |
| 1971 |
} |
| 1972 |
|
| 1973 |
pub fn get_attest(&self, r: &ObjectRef) -> Result<Attest> { |
| 1974 |
Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?) |
| 1975 |
} |
| 1976 |
|
| 1977 |
pub fn put_receipt(&self, r: &Receipt) -> Result<ObjectRef> { |
| 1978 |
self.put(ObjectType::Receipt, &r.encode()) |
| 1979 |
} |
| 1980 |
|
| 1981 |
pub fn get_receipt(&self, r: &ObjectRef) -> Result<Receipt> { |
| 1982 |
Ok(Receipt::decode(&self.get_typed(r, ObjectType::Receipt)?)?) |
| 1983 |
} |
| 1984 |
|
| 1985 |
|
| 1986 |
|
| 1987 |
fn index_add(&self, index: &str, key: &ObjectRef, value: &ObjectRef) -> Result<()> { |
| 1988 |
let entry = format!("{}{}", fanout_prefix(index, key), value.to_hex()); |
| 1989 |
put_new_or_existing(self.backend.as_ref(), &entry, b"") |
| 1990 |
} |
| 1991 |
|
| 1992 |
|
| 1993 |
|
| 1994 |
fn index_get(&self, index: &str, legacy: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> { |
| 1995 |
let mut values: Vec<ObjectRef> = self |
| 1996 |
.backend |
| 1997 |
.list(&fanout_prefix(index, key)) |
| 1998 |
.map_err(backend_err)? |
| 1999 |
.into_iter() |
| 2000 |
.filter_map(|e| e.key.rsplit('/').next().and_then(ObjectRef::from_hex)) |
| 2001 |
.collect(); |
| 2002 |
if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key(legacy, key))? { |
| 2003 |
values.extend( |
| 2004 |
String::from_utf8_lossy(&bytes) |
| 2005 |
.lines() |
| 2006 |
.filter_map(ObjectRef::from_hex), |
| 2007 |
); |
| 2008 |
} |
| 2009 |
values.sort(); |
| 2010 |
values.dedup(); |
| 2011 |
Ok(values) |
| 2012 |
} |
| 2013 |
|
| 2014 |
|
| 2015 |
pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> { |
| 2016 |
self.index_add("seal", change, attest) |
| 2017 |
} |
| 2018 |
|
| 2019 |
|
| 2020 |
pub fn seals_for(&self, change: &ObjectRef) -> Result<Vec<ObjectRef>> { |
| 2021 |
self.index_get("seal", "seals", change) |
| 2022 |
} |
| 2023 |
|
| 2024 |
|
| 2025 |
|
| 2026 |
|
| 2027 |
pub fn put_certificate(&self, c: &crate::cert::SignedCertificate) -> Result<ObjectRef> { |
| 2028 |
let r = self.put(ObjectType::Certificate, &c.cert.encode())?; |
| 2029 |
self.put_signature(&r, &c.sig)?; |
| 2030 |
self.index_add("cert", &c.cert.key_id(), &r)?; |
| 2031 |
Ok(r) |
| 2032 |
} |
| 2033 |
|
| 2034 |
|
| 2035 |
pub fn get_certificate(&self, r: &ObjectRef) -> Result<crate::cert::SignedCertificate> { |
| 2036 |
let cert = crate::cert::Certificate::decode(&self.get_typed(r, ObjectType::Certificate)?)?; |
| 2037 |
let sig = self |
| 2038 |
.signatures(r)? |
| 2039 |
.into_iter() |
| 2040 |
.find(|s| s.key == cert.issuer) |
| 2041 |
.ok_or(StoreError::NotFound(*r))?; |
| 2042 |
Ok(crate::cert::SignedCertificate { cert, sig }) |
| 2043 |
} |
| 2044 |
|
| 2045 |
|
| 2046 |
pub fn certificate_of(&self, key: &ObjectRef) -> Result<Option<ObjectRef>> { |
| 2047 |
Ok(self |
| 2048 |
.index_get("cert", "certs", key)? |
| 2049 |
.into_iter() |
| 2050 |
.find(|r| self.has(r))) |
| 2051 |
} |
| 2052 |
|
| 2053 |
|
| 2054 |
pub(crate) fn index_add_pub( |
| 2055 |
&self, |
| 2056 |
index: &str, |
| 2057 |
key: &ObjectRef, |
| 2058 |
value: &ObjectRef, |
| 2059 |
) -> Result<()> { |
| 2060 |
self.index_add(index, key, value) |
| 2061 |
} |
| 2062 |
|
| 2063 |
|
| 2064 |
|
| 2065 |
pub(crate) fn index_get_pub(&self, index: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> { |
| 2066 |
self.index_get(index, &format!("{index}-lists"), key) |
| 2067 |
} |
| 2068 |
|
| 2069 |
pub fn index_receipt(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> { |
| 2070 |
self.index_add("receipt", state, attest) |
| 2071 |
} |
| 2072 |
|
| 2073 |
|
| 2074 |
pub fn receipts_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> { |
| 2075 |
self.index_get("receipt", "receipts", state) |
| 2076 |
} |
| 2077 |
|
| 2078 |
pub fn index_approval(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> { |
| 2079 |
self.index_add("approval", state, attest) |
| 2080 |
} |
| 2081 |
|
| 2082 |
|
| 2083 |
|
| 2084 |
pub fn approvals_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> { |
| 2085 |
self.index_get("approval", "approvals", state) |
| 2086 |
} |
| 2087 |
|
| 2088 |
|
| 2089 |
pub fn approval_states(&self) -> Result<Vec<ObjectRef>> { |
| 2090 |
self.indexed_states("approval", "approvals") |
| 2091 |
} |
| 2092 |
|
| 2093 |
|
| 2094 |
pub fn receipt_states(&self) -> Result<Vec<ObjectRef>> { |
| 2095 |
self.indexed_states("receipt", "receipts") |
| 2096 |
} |
| 2097 |
|
| 2098 |
|
| 2099 |
|
| 2100 |
fn indexed_states(&self, one: &str, many: &str) -> Result<Vec<ObjectRef>> { |
| 2101 |
let mut out: Vec<ObjectRef> = Vec::new(); |
| 2102 |
for e in self.backend.list(&format!("{one}/")).map_err(backend_err)? { |
| 2103 |
|
| 2104 |
let mut parts = e.key.splitn(4, '/'); |
| 2105 |
let (Some(_), Some(hh), Some(rest), Some(_)) = |
| 2106 |
(parts.next(), parts.next(), parts.next(), parts.next()) |
| 2107 |
else { |
| 2108 |
continue; |
| 2109 |
}; |
| 2110 |
out.extend(ObjectRef::from_hex(&format!("{hh}{rest}"))); |
| 2111 |
} |
| 2112 |
for e in self |
| 2113 |
.backend |
| 2114 |
.list(&format!("{many}/")) |
| 2115 |
.map_err(backend_err)? |
| 2116 |
{ |
| 2117 |
let mut parts = e.key.splitn(3, '/'); |
| 2118 |
let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next()) else { |
| 2119 |
continue; |
| 2120 |
}; |
| 2121 |
out.extend(ObjectRef::from_hex(&format!("{hh}{rest}"))); |
| 2122 |
} |
| 2123 |
out.sort(); |
| 2124 |
out.dedup(); |
| 2125 |
Ok(out) |
| 2126 |
} |
| 2127 |
|
| 2128 |
pub fn put_identity(&self, id: &Identity) -> Result<ObjectRef> { |
| 2129 |
self.put(ObjectType::Identity, &id.encode()) |
| 2130 |
} |
| 2131 |
|
| 2132 |
pub fn get_identity(&self, r: &ObjectRef) -> Result<Identity> { |
| 2133 |
Ok(Identity::decode(&self.get_typed(r, ObjectType::Identity)?)?) |
| 2134 |
} |
| 2135 |
|
| 2136 |
|
| 2137 |
|
| 2138 |
pub fn signatures(&self, r: &ObjectRef) -> Result<Vec<Signature>> { |
| 2139 |
let mut sigs: Vec<Signature> = Vec::new(); |
| 2140 |
for e in self |
| 2141 |
.backend |
| 2142 |
.list(&fanout_prefix("sig", r)) |
| 2143 |
.map_err(backend_err)? |
| 2144 |
{ |
| 2145 |
let bytes = self.backend.get(&e.key).map_err(backend_err)?; |
| 2146 |
sigs.extend(Signature::decode_list(&bytes)?); |
| 2147 |
} |
| 2148 |
|
| 2149 |
|
| 2150 |
if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key("sigs", r))? { |
| 2151 |
for sig in Signature::decode_list(&bytes)? { |
| 2152 |
if !sigs.iter().any(|s| s.key == sig.key) { |
| 2153 |
sigs.push(sig); |
| 2154 |
} |
| 2155 |
} |
| 2156 |
} |
| 2157 |
sigs.sort_by_key(|s| s.key); |
| 2158 |
sigs.dedup_by_key(|s| s.key); |
| 2159 |
Ok(sigs) |
| 2160 |
} |
| 2161 |
|
| 2162 |
|
| 2163 |
|
| 2164 |
|
| 2165 |
pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> { |
| 2166 |
let key = format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex()); |
| 2167 |
let bytes = Signature::encode_list(std::slice::from_ref(sig)); |
| 2168 |
if get_optional(self.backend.as_ref(), &key)?.as_deref() == Some(bytes.as_slice()) { |
| 2169 |
return Ok(()); |
| 2170 |
} |
| 2171 |
delete(self.backend.as_ref(), &key)?; |
| 2172 |
put_new_or_existing(self.backend.as_ref(), &key, &bytes) |
| 2173 |
} |
| 2174 |
|
| 2175 |
|
| 2176 |
|
| 2177 |
|
| 2178 |
|
| 2179 |
|
| 2180 |
|
| 2181 |
|
| 2182 |
pub fn adopt_unfinished_packs( |
| 2183 |
&self, |
| 2184 |
min_age: std::time::Duration, |
| 2185 |
progress: &dyn Fn(&str), |
| 2186 |
) -> Result<(usize, usize)> { |
| 2187 |
use std::io::{BufReader, Read}; |
| 2188 |
let mut packs = 0; |
| 2189 |
let mut objects = 0; |
| 2190 |
let Ok(dir) = fs::read_dir(&self.scratch) else { |
| 2191 |
return Ok((0, 0)); |
| 2192 |
}; |
| 2193 |
for entry in dir.flatten() { |
| 2194 |
let name = entry.file_name().to_string_lossy().to_string(); |
| 2195 |
let Some(id) = name |
| 2196 |
.strip_prefix(".tmp-") |
| 2197 |
.and_then(|n| n.strip_suffix(".pack")) |
| 2198 |
else { |
| 2199 |
continue; |
| 2200 |
}; |
| 2201 |
let path = entry.path(); |
| 2202 |
let meta = fs::metadata(&path)?; |
| 2203 |
let age = meta |
| 2204 |
.modified() |
| 2205 |
.ok() |
| 2206 |
.and_then(|m| m.elapsed().ok()) |
| 2207 |
.unwrap_or_default(); |
| 2208 |
if age < min_age { |
| 2209 |
progress(&format!( |
| 2210 |
"{name}: written to {}s ago, left alone", |
| 2211 |
age.as_secs() |
| 2212 |
)); |
| 2213 |
continue; |
| 2214 |
} |
| 2215 |
let mut f = BufReader::with_capacity(1 << 20, fs::File::open(&path)?); |
| 2216 |
let mut magic = [0u8; 8]; |
| 2217 |
if f.read_exact(&mut magic).is_err() || &magic != PACK_MAGIC { |
| 2218 |
progress(&format!("{name}: not a pack, left alone")); |
| 2219 |
continue; |
| 2220 |
} |
| 2221 |
let mut offset = PACK_MAGIC.len() as u64; |
| 2222 |
let mut index: Vec<(ObjectRef, PackEntry)> = Vec::new(); |
| 2223 |
loop { |
| 2224 |
let mut head = [0u8; 42]; |
| 2225 |
if f.read_exact(&mut head).is_err() { |
| 2226 |
break; |
| 2227 |
} |
| 2228 |
let Some(ty) = ObjectType::ALL.get(head[0] as usize).copied() else { |
| 2229 |
break; |
| 2230 |
}; |
| 2231 |
let flags = head[1]; |
| 2232 |
let stored_len = u32::from_le_bytes(head[2..6].try_into().unwrap()); |
| 2233 |
let raw_len = u32::from_le_bytes(head[6..10].try_into().unwrap()); |
| 2234 |
let r = ObjectRef(head[10..42].try_into().unwrap()); |
| 2235 |
let mut stored = vec![0u8; stored_len as usize]; |
| 2236 |
if flags & FLAG_DICT != 0 || f.read_exact(&mut stored).is_err() { |
| 2237 |
break; |
| 2238 |
} |
| 2239 |
let e = PackEntry { |
| 2240 |
offset: offset + 42, |
| 2241 |
stored_len, |
| 2242 |
raw_len, |
| 2243 |
ty, |
| 2244 |
flags, |
| 2245 |
}; |
| 2246 |
if decode_entry(&r, e, stored, None).is_err() { |
| 2247 |
break; |
| 2248 |
} |
| 2249 |
index.push((r, e)); |
| 2250 |
offset += 42 + stored_len as u64; |
| 2251 |
if index.len().is_multiple_of(200_000) { |
| 2252 |
progress(&format!( |
| 2253 |
"{name}: {} objects checked, {} MiB", |
| 2254 |
index.len(), |
| 2255 |
offset >> 20 |
| 2256 |
)); |
| 2257 |
} |
| 2258 |
} |
| 2259 |
if index.is_empty() { |
| 2260 |
progress(&format!("{name}: no whole object, left alone")); |
| 2261 |
continue; |
| 2262 |
} |
| 2263 |
if offset < meta.len() { |
| 2264 |
progress(&format!( |
| 2265 |
"{name}: {} torn bytes at the end cut off", |
| 2266 |
meta.len() - offset |
| 2267 |
)); |
| 2268 |
fs::OpenOptions::new() |
| 2269 |
.write(true) |
| 2270 |
.open(&path)? |
| 2271 |
.set_len(offset)?; |
| 2272 |
} |
| 2273 |
index.sort_by_key(|(r, _)| *r); |
| 2274 |
index.dedup_by_key(|(r, _)| *r); |
| 2275 |
objects += index.len(); |
| 2276 |
packs += 1; |
| 2277 |
progress(&format!("{name}: {} objects indexed", index.len())); |
| 2278 |
self.finish_pack_files(id, &path, &index)?; |
| 2279 |
} |
| 2280 |
Ok((packs, objects)) |
| 2281 |
} |
| 2282 |
|
| 2283 |
|
| 2284 |
pub fn list(&self) -> io::Result<Vec<ObjectRef>> { |
| 2285 |
let mut out: Vec<ObjectRef> = self |
| 2286 |
.loose_objects() |
| 2287 |
.map_err(|e| io::Error::other(e.to_string()))? |
| 2288 |
.into_iter() |
| 2289 |
.map(|(r, _)| r) |
| 2290 |
.collect(); |
| 2291 |
for p in self.packs.borrow().iter() { |
| 2292 |
out.extend(p.refs.iter().copied()); |
| 2293 |
} |
| 2294 |
out.sort(); |
| 2295 |
out.dedup(); |
| 2296 |
Ok(out) |
| 2297 |
} |
| 2298 |
} |
| 2299 |
|
| 2300 |
|
| 2301 |
|
| 2302 |
fn is_miss(e: &StoreError) -> bool { |
| 2303 |
match e { |
| 2304 |
StoreError::NotFound(_) => true, |
| 2305 |
StoreError::Corrupt(_, why) => why == NO_DICT, |
| 2306 |
_ => false, |
| 2307 |
} |
| 2308 |
} |
| 2309 |
|
| 2310 |
|
| 2311 |
fn parse_loose(data: &[u8], r: &ObjectRef) -> Result<Object> { |
| 2312 |
let nul = data |
| 2313 |
.iter() |
| 2314 |
.position(|&b| b == 0) |
| 2315 |
.ok_or_else(|| StoreError::Corrupt(*r, "no header".into()))?; |
| 2316 |
let head = std::str::from_utf8(&data[..nul]) |
| 2317 |
.map_err(|_| StoreError::Corrupt(*r, "header not UTF-8".into()))?; |
| 2318 |
let mut parts = head.split(' '); |
| 2319 |
let (Some("oxy1"), Some(ty), Some(len), None) = |
| 2320 |
(parts.next(), parts.next(), parts.next(), parts.next()) |
| 2321 |
else { |
| 2322 |
return Err(StoreError::Corrupt(*r, format!("bad header {head:?}"))); |
| 2323 |
}; |
| 2324 |
let ty = ObjectType::from_name(ty) |
| 2325 |
.ok_or_else(|| StoreError::Corrupt(*r, format!("unknown type {ty}")))?; |
| 2326 |
let len: usize = len |
| 2327 |
.parse() |
| 2328 |
.map_err(|_| StoreError::Corrupt(*r, "bad length".into()))?; |
| 2329 |
let body = &data[nul + 1..]; |
| 2330 |
if body.len() != len { |
| 2331 |
return Err(StoreError::Corrupt( |
| 2332 |
*r, |
| 2333 |
format!("length {} != header {len}", body.len()), |
| 2334 |
)); |
| 2335 |
} |
| 2336 |
if hash_object(ty, body) != *r { |
| 2337 |
return Err(StoreError::Corrupt(*r, "hash mismatch".into())); |
| 2338 |
} |
| 2339 |
Ok((ty, body.to_vec())) |
| 2340 |
} |
| 2341 |
|
| 2342 |
|
| 2343 |
|
| 2344 |
#[cfg(test)] |
| 2345 |
fn files_on_disk() -> bool { |
| 2346 |
std::env::var_os("OXY_STORE_BACKEND").is_none() |
| 2347 |
} |
| 2348 |
|
| 2349 |
#[cfg(test)] |
| 2350 |
mod tests { |
| 2351 |
use super::*; |
| 2352 |
|
| 2353 |
|
| 2354 |
|
| 2355 |
fn each_store(dir: &Path, f: impl Fn(&str, Store)) { |
| 2356 |
f("directory", Store::open(dir.join("objects")).unwrap()); |
| 2357 |
f( |
| 2358 |
"memory", |
| 2359 |
Store::open_on(Arc::new(MemoryStore::new()), dir.join("scratch")).unwrap(), |
| 2360 |
); |
| 2361 |
} |
| 2362 |
|
| 2363 |
#[test] |
| 2364 |
fn put_get_verify() { |
| 2365 |
if !super::files_on_disk() { |
| 2366 |
return; |
| 2367 |
} |
| 2368 |
let dir = tempfile::tempdir().unwrap(); |
| 2369 |
let s = Store::open(dir.path().join("objects")).unwrap(); |
| 2370 |
let r = s.put(ObjectType::Chunk, b"hello").unwrap(); |
| 2371 |
assert!(s.has(&r)); |
| 2372 |
assert_eq!(s.get(&r).unwrap(), (ObjectType::Chunk, b"hello".to_vec())); |
| 2373 |
assert_eq!(s.put(ObjectType::Chunk, b"hello").unwrap(), r); |
| 2374 |
assert_eq!(s.list().unwrap(), vec![r]); |
| 2375 |
assert!(matches!( |
| 2376 |
s.get_typed(&r, ObjectType::Blob), |
| 2377 |
Err(StoreError::WrongType(..)) |
| 2378 |
)); |
| 2379 |
assert!(matches!( |
| 2380 |
s.get(&ObjectRef([9; 32])), |
| 2381 |
Err(StoreError::NotFound(_)) |
| 2382 |
)); |
| 2383 |
|
| 2384 |
|
| 2385 |
let path = s.path_for(&r); |
| 2386 |
let mut data = fs::read(&path).unwrap(); |
| 2387 |
let last = data.len() - 1; |
| 2388 |
data[last] ^= 1; |
| 2389 |
fs::write(&path, data).unwrap(); |
| 2390 |
assert!(matches!(s.get(&r), Err(StoreError::Corrupt(..)))); |
| 2391 |
} |
| 2392 |
|
| 2393 |
#[test] |
| 2394 |
fn packing_keeps_every_object_readable() { |
| 2395 |
if !super::files_on_disk() { |
| 2396 |
return; |
| 2397 |
} |
| 2398 |
let dir = tempfile::tempdir().unwrap(); |
| 2399 |
let s = Store::open(dir.path()).unwrap(); |
| 2400 |
let mut refs = Vec::new(); |
| 2401 |
for i in 0..300u32 { |
| 2402 |
let body = format!("object {i} ").repeat(1 + (i % 7) as usize); |
| 2403 |
refs.push((s.put(ObjectType::Chunk, body.as_bytes()).unwrap(), body)); |
| 2404 |
} |
| 2405 |
let c = crate::objects::tests::sample_change(); |
| 2406 |
let cr = s.put_change(&c).unwrap(); |
| 2407 |
assert_eq!(s.loose_count().unwrap(), 301); |
| 2408 |
let report = s.pack_loose().unwrap(); |
| 2409 |
assert_eq!(report.objects, 301); |
| 2410 |
assert!(report.packed_bytes < report.raw_bytes, "{report:?}"); |
| 2411 |
assert_eq!(s.loose_count().unwrap(), 0); |
| 2412 |
assert_eq!(s.pack_count(), 1); |
| 2413 |
for (r, body) in &refs { |
| 2414 |
assert!(s.has(r)); |
| 2415 |
assert_eq!( |
| 2416 |
s.get(r).unwrap(), |
| 2417 |
(ObjectType::Chunk, body.as_bytes().to_vec()) |
| 2418 |
); |
| 2419 |
} |
| 2420 |
assert_eq!(s.get_change(&cr).unwrap(), c); |
| 2421 |
assert_eq!(s.list().unwrap().len(), 301); |
| 2422 |
|
| 2423 |
|
| 2424 |
assert_eq!( |
| 2425 |
s.put(ObjectType::Chunk, refs[0].1.as_bytes()).unwrap(), |
| 2426 |
refs[0].0 |
| 2427 |
); |
| 2428 |
assert_eq!(s.loose_count().unwrap(), 0); |
| 2429 |
let extra = s.put(ObjectType::Chunk, b"after packing").unwrap(); |
| 2430 |
s.pack_loose().unwrap(); |
| 2431 |
assert_eq!(s.pack_count(), 2); |
| 2432 |
assert_eq!(s.get(&extra).unwrap().1, b"after packing"); |
| 2433 |
|
| 2434 |
let s2 = Store::open(dir.path()).unwrap(); |
| 2435 |
assert_eq!(s2.pack_count(), 2); |
| 2436 |
assert_eq!(s2.get(&refs[5].0).unwrap().1, refs[5].1.as_bytes()); |
| 2437 |
assert!(matches!( |
| 2438 |
s2.get(&ObjectRef([9; 32])), |
| 2439 |
Err(StoreError::NotFound(_)) |
| 2440 |
)); |
| 2441 |
|
| 2442 |
let target = refs[5].0; |
| 2443 |
let (pack_path, offset) = { |
| 2444 |
let packs = s2.packs.borrow(); |
| 2445 |
let (i, e) = s2.find_packed(&target).unwrap(); |
| 2446 |
(dir.path().join(&packs[i].key), e.offset as usize) |
| 2447 |
}; |
| 2448 |
let mut data = fs::read(&pack_path).unwrap(); |
| 2449 |
data[offset] ^= 0xff; |
| 2450 |
fs::write(&pack_path, data).unwrap(); |
| 2451 |
let s3 = Store::open(dir.path()).unwrap(); |
| 2452 |
assert!(matches!(s3.get(&target), Err(StoreError::Corrupt(..)))); |
| 2453 |
assert_eq!( |
| 2454 |
s3.get(&refs[6].0).unwrap().1, |
| 2455 |
refs[6].1.as_bytes(), |
| 2456 |
"other objects are unaffected" |
| 2457 |
); |
| 2458 |
} |
| 2459 |
|
| 2460 |
#[test] |
| 2461 |
fn a_full_repack_trains_a_dictionary_and_reads_back() { |
| 2462 |
if !super::files_on_disk() { |
| 2463 |
return; |
| 2464 |
} |
| 2465 |
let dir = tempfile::tempdir().unwrap(); |
| 2466 |
let s = Store::open(dir.path()).unwrap(); |
| 2467 |
|
| 2468 |
let bodies: Vec<Vec<u8>> = (0..1500) |
| 2469 |
.map(|i| { |
| 2470 |
format!( |
| 2471 |
"// SPDX-License-Identifier: GPL-2.0\n#include <linux/module.h>\n\ |
| 2472 |
static int probe_{i}(struct device *dev)\n{{\n\treturn {i};\n}}\n" |
| 2473 |
) |
| 2474 |
.into_bytes() |
| 2475 |
}) |
| 2476 |
.collect(); |
| 2477 |
let refs: Vec<ObjectRef> = bodies |
| 2478 |
.iter() |
| 2479 |
.map(|b| s.put(ObjectType::Chunk, b).unwrap()) |
| 2480 |
.collect(); |
| 2481 |
s.repack_all(REPACK_ZSTD_LEVEL).unwrap(); |
| 2482 |
let dicts = fs::read_dir(dir.path()) |
| 2483 |
.unwrap() |
| 2484 |
.flatten() |
| 2485 |
.filter(|e| e.path().extension().is_some_and(|x| x == "dict")) |
| 2486 |
.count(); |
| 2487 |
assert_eq!(dicts, 1, "the pack has a dictionary"); |
| 2488 |
let with_dict = refs |
| 2489 |
.iter() |
| 2490 |
.filter(|r| s.find_packed(r).unwrap().1.flags & FLAG_DICT != 0) |
| 2491 |
.count(); |
| 2492 |
assert!(with_dict > 1000, "most use it: {with_dict}"); |
| 2493 |
let again = Store::open(dir.path()).unwrap(); |
| 2494 |
for (r, b) in refs.iter().zip(&bodies) { |
| 2495 |
assert_eq!(&again.get(r).unwrap().1, b); |
| 2496 |
} |
| 2497 |
let packs = |dir: &Path| { |
| 2498 |
let mut v: Vec<String> = fs::read_dir(dir) |
| 2499 |
.unwrap() |
| 2500 |
.flatten() |
| 2501 |
.map(|e| e.file_name().to_string_lossy().into_owned()) |
| 2502 |
.filter(|n| n.starts_with("pack-")) |
| 2503 |
.collect(); |
| 2504 |
v.sort(); |
| 2505 |
v |
| 2506 |
}; |
| 2507 |
|
| 2508 |
|
| 2509 |
let before = packs(dir.path()); |
| 2510 |
let (report, merged) = again.repack_all(REPACK_ZSTD_LEVEL).unwrap(); |
| 2511 |
assert_eq!((report.objects, merged), (0, 0)); |
| 2512 |
assert_eq!(packs(dir.path()), before); |
| 2513 |
|
| 2514 |
|
| 2515 |
let extra = again.put(ObjectType::Chunk, b"one more\n").unwrap(); |
| 2516 |
again.repack_all(REPACK_ZSTD_LEVEL).unwrap(); |
| 2517 |
let files = packs(dir.path()); |
| 2518 |
assert_eq!( |
| 2519 |
files.len(), |
| 2520 |
7, |
| 2521 |
"two of pack, index and dictionary, and the retired record: {files:?}" |
| 2522 |
); |
| 2523 |
assert_eq!(again.pack_count(), 1, "the old pack is not read"); |
| 2524 |
assert_eq!(&again.get(&refs[7]).unwrap().1, &bodies[7]); |
| 2525 |
assert_eq!(again.get(&extra).unwrap().1, b"one more\n"); |
| 2526 |
again |
| 2527 |
.expire_retired(std::time::Duration::ZERO, &mut silent()) |
| 2528 |
.unwrap(); |
| 2529 |
let files = packs(dir.path()); |
| 2530 |
assert_eq!(files.len(), 4, "one pack, index, dictionary: {files:?}"); |
| 2531 |
assert!(before |
| 2532 |
.iter() |
| 2533 |
.filter(|f| *f != RETIRED_KEY) |
| 2534 |
.all(|f| !files.contains(f))); |
| 2535 |
assert_eq!(&again.get(&refs[8]).unwrap().1, &bodies[8]); |
| 2536 |
} |
| 2537 |
|
| 2538 |
#[test] |
| 2539 |
fn nocompress_chunks_pack_raw() { |
| 2540 |
let dir = tempfile::tempdir().unwrap(); |
| 2541 |
let s = Store::open(dir.path()).unwrap(); |
| 2542 |
let body = vec![b'a'; 10_000]; |
| 2543 |
let raw = s.put_chunk_nocompress(&body).unwrap(); |
| 2544 |
let compressible = s.put(ObjectType::Chunk, &vec![b'b'; 10_000]).unwrap(); |
| 2545 |
s.pack_loose().unwrap(); |
| 2546 |
let packs = s.packs.borrow(); |
| 2547 |
let (i, e) = s.find_packed(&raw).unwrap(); |
| 2548 |
assert_eq!((e.flags, e.stored_len as usize), (0, 10_000)); |
| 2549 |
let (_, c) = s.find_packed(&compressible).unwrap(); |
| 2550 |
assert_eq!(c.flags, FLAG_ZSTD); |
| 2551 |
assert!(c.stored_len < 10_000); |
| 2552 |
assert_eq!(packs[i].fetch(s.backend.as_ref(), &raw, e).unwrap().1, body); |
| 2553 |
} |
| 2554 |
|
| 2555 |
#[test] |
| 2556 |
fn many_packs_merge_into_one() { |
| 2557 |
let dir = tempfile::tempdir().unwrap(); |
| 2558 |
each_store(dir.path(), |which, s| { |
| 2559 |
let mut refs = Vec::new(); |
| 2560 |
for i in 0..(MAX_PACKS + 2) { |
| 2561 |
refs.push( |
| 2562 |
s.put(ObjectType::Chunk, format!("pack {i}").as_bytes()) |
| 2563 |
.unwrap(), |
| 2564 |
); |
| 2565 |
s.pack_loose().unwrap(); |
| 2566 |
} |
| 2567 |
assert_eq!(s.pack_count(), MAX_PACKS + 2, "{which}"); |
| 2568 |
assert_eq!(s.merge_packs().unwrap(), MAX_PACKS + 2); |
| 2569 |
assert_eq!(s.pack_count(), 1); |
| 2570 |
assert_eq!(s.merge_packs().unwrap(), 0); |
| 2571 |
for (i, r) in refs.iter().enumerate() { |
| 2572 |
assert_eq!(s.get(r).unwrap().1, format!("pack {i}").as_bytes()); |
| 2573 |
} |
| 2574 |
let again = Store::open_on(s.backend().clone(), s.dir()).unwrap(); |
| 2575 |
assert_eq!(again.pack_count(), 1, "{which}: reopened"); |
| 2576 |
}); |
| 2577 |
} |
| 2578 |
|
| 2579 |
#[test] |
| 2580 |
fn prune_removes_unreachable_loose_and_packed_objects() { |
| 2581 |
use std::collections::HashSet; |
| 2582 |
let dir = tempfile::tempdir().unwrap(); |
| 2583 |
each_store(dir.path(), |which, s| { |
| 2584 |
let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap(); |
| 2585 |
let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap(); |
| 2586 |
s.pack_loose().unwrap(); |
| 2587 |
let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap(); |
| 2588 |
let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap(); |
| 2589 |
let sk = crate::identity::SecretKey { |
| 2590 |
ed_seed: [1; 32], |
| 2591 |
ml_seed: [2; 32], |
| 2592 |
}; |
| 2593 |
let id = sk.identity("a", "b", 0); |
| 2594 |
let key = s.put_identity(&id).unwrap(); |
| 2595 |
s.put_signature( |
| 2596 |
&drop_b, |
| 2597 |
&sk.sign(crate::identity::Domain::Change, &key, &drop_b), |
| 2598 |
) |
| 2599 |
.unwrap(); |
| 2600 |
s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap(); |
| 2601 |
let keep: HashSet<ObjectRef> = [keep_a, keep_b, key].into_iter().collect(); |
| 2602 |
|
| 2603 |
let r = s |
| 2604 |
.prune(&keep, std::time::Duration::from_secs(3600)) |
| 2605 |
.unwrap(); |
| 2606 |
assert_eq!((r.objects, r.deferred), (0, 2), "{which}"); |
| 2607 |
let r = s.prune(&keep, std::time::Duration::ZERO).unwrap(); |
| 2608 |
assert_eq!(r.objects, 2, "{which}: {r:?}"); |
| 2609 |
assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key)); |
| 2610 |
assert!(!s.has(&drop_a) && !s.has(&drop_b)); |
| 2611 |
assert!(s.signatures(&drop_b).unwrap().is_empty()); |
| 2612 |
assert!(s.seals_for(&drop_a).unwrap().is_empty()); |
| 2613 |
assert_eq!(s.pack_count(), 1); |
| 2614 |
assert_eq!(s.get(&keep_a).unwrap().1, b"keep a"); |
| 2615 |
assert_eq!(s.list().unwrap().len(), 3); |
| 2616 |
}); |
| 2617 |
} |
| 2618 |
|
| 2619 |
#[test] |
| 2620 |
fn signatures_are_stored_beside_objects() { |
| 2621 |
use crate::identity::{Domain, SecretKey}; |
| 2622 |
let dir = tempfile::tempdir().unwrap(); |
| 2623 |
each_store(dir.path(), |which, s| { |
| 2624 |
let r = s.put(ObjectType::Chunk, b"x").unwrap(); |
| 2625 |
assert!(s.signatures(&r).unwrap().is_empty()); |
| 2626 |
let sk = SecretKey { |
| 2627 |
ed_seed: [1; 32], |
| 2628 |
ml_seed: [2; 32], |
| 2629 |
}; |
| 2630 |
let id = sk.identity("a", "b", 0); |
| 2631 |
let key = s.put_identity(&id).unwrap(); |
| 2632 |
assert_eq!(s.get_identity(&key).unwrap(), id); |
| 2633 |
let sig = sk.sign(Domain::Change, &key, &r); |
| 2634 |
s.put_signature(&r, &sig).unwrap(); |
| 2635 |
s.put_signature(&r, &sig).unwrap(); |
| 2636 |
assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()], "{which}"); |
| 2637 |
|
| 2638 |
let sk2 = SecretKey { |
| 2639 |
ed_seed: [3; 32], |
| 2640 |
ml_seed: [4; 32], |
| 2641 |
}; |
| 2642 |
let key2 = s.put_identity(&sk2.identity("c", "d", 0)).unwrap(); |
| 2643 |
let sig2 = sk2.sign(Domain::Change, &key2, &r); |
| 2644 |
s.put_signature(&r, &sig2).unwrap(); |
| 2645 |
let mut both = vec![sig.clone(), sig2.clone()]; |
| 2646 |
both.sort_by_key(|s| s.key); |
| 2647 |
assert_eq!(s.signatures(&r).unwrap(), both); |
| 2648 |
|
| 2649 |
assert_eq!(s.get(&r).unwrap().1, b"x"); |
| 2650 |
|
| 2651 |
s.index_seal(&r, &ObjectRef([5; 32])).unwrap(); |
| 2652 |
s.index_seal(&r, &ObjectRef([4; 32])).unwrap(); |
| 2653 |
s.index_seal(&r, &ObjectRef([5; 32])).unwrap(); |
| 2654 |
assert_eq!( |
| 2655 |
s.seals_for(&r).unwrap(), |
| 2656 |
vec![ObjectRef([4; 32]), ObjectRef([5; 32])] |
| 2657 |
); |
| 2658 |
let st = ObjectRef([6; 32]); |
| 2659 |
s.index_receipt(&st, &ObjectRef([7; 32])).unwrap(); |
| 2660 |
assert_eq!(s.receipts_for(&st).unwrap(), vec![ObjectRef([7; 32])]); |
| 2661 |
assert_eq!(s.receipt_states().unwrap(), vec![st]); |
| 2662 |
}); |
| 2663 |
} |
| 2664 |
|
| 2665 |
|
| 2666 |
|
| 2667 |
|
| 2668 |
#[test] |
| 2669 |
fn an_older_store_s_sidecar_lists_are_still_read() { |
| 2670 |
if !super::files_on_disk() { |
| 2671 |
return; |
| 2672 |
} |
| 2673 |
use crate::identity::{Domain, SecretKey}; |
| 2674 |
let dir = tempfile::tempdir().unwrap(); |
| 2675 |
let s = Store::open(dir.path()).unwrap(); |
| 2676 |
let r = s.put(ObjectType::Chunk, b"old").unwrap(); |
| 2677 |
let h = r.to_hex(); |
| 2678 |
let sk = SecretKey { |
| 2679 |
ed_seed: [1; 32], |
| 2680 |
ml_seed: [2; 32], |
| 2681 |
}; |
| 2682 |
let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap(); |
| 2683 |
let sig = sk.sign(Domain::Change, &key, &r); |
| 2684 |
let legacy = dir.path().join("sigs").join(&h[..2]); |
| 2685 |
fs::create_dir_all(&legacy).unwrap(); |
| 2686 |
fs::write( |
| 2687 |
legacy.join(&h[2..]), |
| 2688 |
Signature::encode_list(std::slice::from_ref(&sig)), |
| 2689 |
) |
| 2690 |
.unwrap(); |
| 2691 |
let seals = dir.path().join("seals").join(&h[..2]); |
| 2692 |
fs::create_dir_all(&seals).unwrap(); |
| 2693 |
fs::write( |
| 2694 |
seals.join(&h[2..]), |
| 2695 |
format!("{}\n", ObjectRef([9; 32]).to_hex()), |
| 2696 |
) |
| 2697 |
.unwrap(); |
| 2698 |
let receipts = dir.path().join("receipts").join(&h[..2]); |
| 2699 |
fs::create_dir_all(&receipts).unwrap(); |
| 2700 |
fs::write( |
| 2701 |
receipts.join(&h[2..]), |
| 2702 |
format!("{}\n", ObjectRef([8; 32]).to_hex()), |
| 2703 |
) |
| 2704 |
.unwrap(); |
| 2705 |
|
| 2706 |
assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]); |
| 2707 |
assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]); |
| 2708 |
assert_eq!(s.receipts_for(&r).unwrap(), vec![ObjectRef([8; 32])]); |
| 2709 |
assert_eq!(s.receipt_states().unwrap(), vec![r]); |
| 2710 |
|
| 2711 |
s.put_signature(&r, &sig).unwrap(); |
| 2712 |
s.index_seal(&r, &ObjectRef([1; 32])).unwrap(); |
| 2713 |
assert_eq!(s.signatures(&r).unwrap().len(), 1); |
| 2714 |
assert_eq!( |
| 2715 |
s.seals_for(&r).unwrap(), |
| 2716 |
vec![ObjectRef([1; 32]), ObjectRef([9; 32])] |
| 2717 |
); |
| 2718 |
|
| 2719 |
let r2 = s.put(ObjectType::Chunk, b"kept").unwrap(); |
| 2720 |
let keep = [r2, key].into_iter().collect(); |
| 2721 |
s.prune(&keep, std::time::Duration::ZERO).unwrap(); |
| 2722 |
assert!(s.signatures(&r).unwrap().is_empty()); |
| 2723 |
assert!(s.seals_for(&r).unwrap().is_empty()); |
| 2724 |
assert!(s.receipt_states().unwrap().is_empty()); |
| 2725 |
} |
| 2726 |
|
| 2727 |
#[test] |
| 2728 |
fn change_round_trip_through_store() { |
| 2729 |
let dir = tempfile::tempdir().unwrap(); |
| 2730 |
each_store(dir.path(), |which, s| { |
| 2731 |
let c = crate::objects::tests::sample_change(); |
| 2732 |
let r = s.put_change(&c).unwrap(); |
| 2733 |
assert_eq!(r, c.hash(), "{which}"); |
| 2734 |
assert_eq!(s.get_change(&r).unwrap(), c); |
| 2735 |
assert_eq!(s.get_change_meta(&r).unwrap().message, c.message); |
| 2736 |
}); |
| 2737 |
} |
| 2738 |
|
| 2739 |
fn xorshift(state: &mut u64) -> u64 { |
| 2740 |
*state ^= *state << 13; |
| 2741 |
*state ^= *state >> 7; |
| 2742 |
*state ^= *state << 17; |
| 2743 |
*state |
| 2744 |
} |
| 2745 |
|
| 2746 |
|
| 2747 |
|
| 2748 |
|
| 2749 |
#[test] |
| 2750 |
fn corrupted_packs_and_indexes_fail_cleanly() { |
| 2751 |
if !super::files_on_disk() { |
| 2752 |
return; |
| 2753 |
} |
| 2754 |
let dir = tempfile::tempdir().unwrap(); |
| 2755 |
let objects = dir.path().join("objects"); |
| 2756 |
let s = Store::open(&objects).unwrap(); |
| 2757 |
let mut refs = Vec::new(); |
| 2758 |
let mut seed = 0x9e3779b97f4a7c15u64; |
| 2759 |
for i in 0..40u32 { |
| 2760 |
let len = (xorshift(&mut seed) % 5000) as usize; |
| 2761 |
let mut body = vec![0u8; len]; |
| 2762 |
for b in body.iter_mut() { |
| 2763 |
*b = if i % 2 == 0 { |
| 2764 |
(xorshift(&mut seed) & 0xff) as u8 |
| 2765 |
} else { |
| 2766 |
b'a' + (i % 26) as u8 |
| 2767 |
}; |
| 2768 |
} |
| 2769 |
refs.push(s.put(ObjectType::Chunk, &body).unwrap()); |
| 2770 |
} |
| 2771 |
s.pack_loose().unwrap(); |
| 2772 |
let pack = fs::read_dir(&objects) |
| 2773 |
.unwrap() |
| 2774 |
.filter_map(|e| e.ok().map(|e| e.path())) |
| 2775 |
.find(|p| p.extension().is_some_and(|x| x == "pack")) |
| 2776 |
.unwrap(); |
| 2777 |
let idx = pack.with_extension("idx"); |
| 2778 |
let pristine_pack = fs::read(&pack).unwrap(); |
| 2779 |
let pristine_idx = fs::read(&idx).unwrap(); |
| 2780 |
|
| 2781 |
for round in 0..300u32 { |
| 2782 |
let (path, pristine) = if round % 2 == 0 { |
| 2783 |
(&pack, &pristine_pack) |
| 2784 |
} else { |
| 2785 |
(&idx, &pristine_idx) |
| 2786 |
}; |
| 2787 |
let mut data = pristine.clone(); |
| 2788 |
match xorshift(&mut seed) % 4 { |
| 2789 |
0 => { |
| 2790 |
let at = (xorshift(&mut seed) as usize) % data.len(); |
| 2791 |
data[at] ^= 1 << (xorshift(&mut seed) % 8); |
| 2792 |
} |
| 2793 |
1 => { |
| 2794 |
let at = (xorshift(&mut seed) as usize) % data.len(); |
| 2795 |
data[at] = (xorshift(&mut seed) & 0xff) as u8; |
| 2796 |
} |
| 2797 |
2 => { |
| 2798 |
let keep = (xorshift(&mut seed) as usize) % data.len(); |
| 2799 |
data.truncate(keep); |
| 2800 |
} |
| 2801 |
_ => { |
| 2802 |
let at = (xorshift(&mut seed) as usize) % data.len(); |
| 2803 |
let n = ((xorshift(&mut seed) % 8) + 1) as usize; |
| 2804 |
for k in 0..n.min(data.len() - at) { |
| 2805 |
data[at + k] = 0xff; |
| 2806 |
} |
| 2807 |
} |
| 2808 |
} |
| 2809 |
fs::write(path, &data).unwrap(); |
| 2810 |
|
| 2811 |
|
| 2812 |
if let Ok(s2) = Store::open(&objects) { |
| 2813 |
for r in &refs { |
| 2814 |
match s2.get(r) { |
| 2815 |
Ok((ty, body)) => assert_eq!( |
| 2816 |
hash_object(ty, &body), |
| 2817 |
*r, |
| 2818 |
"a returned object always matches its reference" |
| 2819 |
), |
| 2820 |
Err(StoreError::Corrupt(..)) |
| 2821 |
| Err(StoreError::NotFound(_)) |
| 2822 |
| Err(StoreError::Io(_)) => {} |
| 2823 |
Err(e) => panic!("unexpected error {e}"), |
| 2824 |
} |
| 2825 |
} |
| 2826 |
} |
| 2827 |
fs::write(path, pristine).unwrap(); |
| 2828 |
} |
| 2829 |
|
| 2830 |
let s3 = Store::open(&objects).unwrap(); |
| 2831 |
for r in &refs { |
| 2832 |
let (ty, body) = s3.get(r).unwrap(); |
| 2833 |
assert_eq!(hash_object(ty, &body), *r); |
| 2834 |
} |
| 2835 |
} |
| 2836 |
|
| 2837 |
|
| 2838 |
|
| 2839 |
#[test] |
| 2840 |
fn batches_write_loose_or_one_pack() { |
| 2841 |
let dir = tempfile::tempdir().unwrap(); |
| 2842 |
each_store(dir.path(), |which, s| { |
| 2843 |
let outer = s.begin_batch().unwrap(); |
| 2844 |
let a = s.put(ObjectType::Chunk, b"small a").unwrap(); |
| 2845 |
{ |
| 2846 |
let inner = s.begin_batch().unwrap(); |
| 2847 |
let b = s.put(ObjectType::Chunk, b"small b").unwrap(); |
| 2848 |
assert!(s.has(&b)); |
| 2849 |
assert_eq!(s.get(&b).unwrap().1, b"small b"); |
| 2850 |
inner.finish().unwrap(); |
| 2851 |
assert_eq!( |
| 2852 |
s.loose_count().unwrap(), |
| 2853 |
0, |
| 2854 |
"{which}: the inner batch defers to the outer one" |
| 2855 |
); |
| 2856 |
} |
| 2857 |
assert_eq!(s.get(&a).unwrap().1, b"small a"); |
| 2858 |
assert_eq!( |
| 2859 |
s.put(ObjectType::Chunk, b"small a").unwrap(), |
| 2860 |
a, |
| 2861 |
"idempotent while pending" |
| 2862 |
); |
| 2863 |
outer.finish().unwrap(); |
| 2864 |
assert_eq!(s.loose_count().unwrap(), 2); |
| 2865 |
assert_eq!(s.pack_count(), 0); |
| 2866 |
assert_eq!(s.get(&a).unwrap().1, b"small a"); |
| 2867 |
|
| 2868 |
|
| 2869 |
let big = s.begin_batch().unwrap(); |
| 2870 |
let mut refs = Vec::new(); |
| 2871 |
let mut seed = 7u64; |
| 2872 |
for i in 0..300u32 { |
| 2873 |
let mut body = vec![b'z'; 40_000]; |
| 2874 |
if i % 3 == 0 { |
| 2875 |
for b in body.iter_mut() { |
| 2876 |
*b = (xorshift(&mut seed) & 0xff) as u8; |
| 2877 |
} |
| 2878 |
} |
| 2879 |
body[..4].copy_from_slice(&i.to_le_bytes()); |
| 2880 |
let r = if i % 5 == 0 { |
| 2881 |
s.put_chunk_nocompress(&body).unwrap() |
| 2882 |
} else { |
| 2883 |
s.put(ObjectType::Chunk, &body).unwrap() |
| 2884 |
}; |
| 2885 |
assert_eq!( |
| 2886 |
s.get(&r).unwrap().1, |
| 2887 |
body, |
| 2888 |
"readable while pending (object {i})" |
| 2889 |
); |
| 2890 |
refs.push((r, body)); |
| 2891 |
} |
| 2892 |
assert!(s.has(&refs[0].0)); |
| 2893 |
big.finish().unwrap(); |
| 2894 |
assert_eq!(s.pack_count(), 1); |
| 2895 |
assert_eq!( |
| 2896 |
s.loose_count().unwrap(), |
| 2897 |
2, |
| 2898 |
"nothing from the big batch is loose" |
| 2899 |
); |
| 2900 |
assert!(fs::read_dir(s.dir()) |
| 2901 |
.unwrap() |
| 2902 |
.flatten() |
| 2903 |
.all(|e| !e.file_name().to_string_lossy().contains("tmp-"))); |
| 2904 |
let s2 = Store::open_on(s.backend().clone(), s.dir()).unwrap(); |
| 2905 |
for (r, body) in &refs { |
| 2906 |
assert_eq!(s2.get(r).unwrap().1, *body); |
| 2907 |
} |
| 2908 |
|
| 2909 |
{ |
| 2910 |
let _b = s2.begin_batch().unwrap(); |
| 2911 |
s2.put(ObjectType::Chunk, b"dropped").unwrap(); |
| 2912 |
} |
| 2913 |
assert_eq!( |
| 2914 |
s2.get(&hash_object(ObjectType::Chunk, b"dropped")) |
| 2915 |
.unwrap() |
| 2916 |
.1, |
| 2917 |
b"dropped" |
| 2918 |
); |
| 2919 |
}); |
| 2920 |
} |
| 2921 |
|
| 2922 |
#[test] |
| 2923 |
fn get_many_reads_loose_and_packed_objects_in_order() { |
| 2924 |
let dir = tempfile::tempdir().unwrap(); |
| 2925 |
each_store(dir.path(), |which, s| { |
| 2926 |
let packed: Vec<ObjectRef> = (0..20) |
| 2927 |
.map(|i| { |
| 2928 |
s.put(ObjectType::Chunk, format!("packed {i}").as_bytes()) |
| 2929 |
.unwrap() |
| 2930 |
}) |
| 2931 |
.collect(); |
| 2932 |
s.pack_loose().unwrap(); |
| 2933 |
let loose: Vec<ObjectRef> = (0..20) |
| 2934 |
.map(|i| { |
| 2935 |
s.put(ObjectType::Chunk, format!("loose {i}").as_bytes()) |
| 2936 |
.unwrap() |
| 2937 |
}) |
| 2938 |
.collect(); |
| 2939 |
let missing = hash_object(ObjectType::Chunk, b"never stored"); |
| 2940 |
let mut refs: Vec<ObjectRef> = Vec::new(); |
| 2941 |
for i in 0..20 { |
| 2942 |
refs.push(loose[i]); |
| 2943 |
refs.push(packed[i]); |
| 2944 |
} |
| 2945 |
refs.push(missing); |
| 2946 |
let got = s.get_many(&refs); |
| 2947 |
assert_eq!(got.len(), refs.len()); |
| 2948 |
for (i, r) in refs.iter().enumerate().take(40) { |
| 2949 |
let want = if i % 2 == 0 { |
| 2950 |
format!("loose {}", i / 2) |
| 2951 |
} else { |
| 2952 |
format!("packed {}", i / 2) |
| 2953 |
}; |
| 2954 |
let (ty, body) = got[i].as_ref().unwrap(); |
| 2955 |
assert_eq!( |
| 2956 |
(*ty, body.as_slice()), |
| 2957 |
(ObjectType::Chunk, want.as_bytes()), |
| 2958 |
"{which}: {r:?}" |
| 2959 |
); |
| 2960 |
} |
| 2961 |
assert!(matches!(got[40], Err(StoreError::NotFound(_)))); |
| 2962 |
}); |
| 2963 |
} |
| 2964 |
|
| 2965 |
|
| 2966 |
|
| 2967 |
|
| 2968 |
#[test] |
| 2969 |
fn packed_objects_cost_no_loose_lookups() { |
| 2970 |
use std::sync::atomic::{AtomicUsize, Ordering}; |
| 2971 |
struct Counting(MemoryStore, AtomicUsize); |
| 2972 |
impl crate::strata::Store for Counting { |
| 2973 |
fn describe(&self) -> String { |
| 2974 |
"counting".into() |
| 2975 |
} |
| 2976 |
fn has(&self, k: &str) -> crate::strata::Result<bool> { |
| 2977 |
self.0.has(k) |
| 2978 |
} |
| 2979 |
fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> { |
| 2980 |
if k.starts_with("loose/") { |
| 2981 |
self.1.fetch_add(1, Ordering::Relaxed); |
| 2982 |
} |
| 2983 |
self.0.get(k) |
| 2984 |
} |
| 2985 |
fn get_range( |
| 2986 |
&self, |
| 2987 |
k: &str, |
| 2988 |
r: std::ops::Range<u64>, |
| 2989 |
) -> crate::strata::Result<Vec<u8>> { |
| 2990 |
self.0.get_range(k, r) |
| 2991 |
} |
| 2992 |
fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> { |
| 2993 |
self.0.put_new(k, b) |
| 2994 |
} |
| 2995 |
fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> { |
| 2996 |
self.0.list(p) |
| 2997 |
} |
| 2998 |
fn delete(&self, k: &str) -> crate::strata::Result<()> { |
| 2999 |
self.0.delete(k) |
| 3000 |
} |
| 3001 |
fn read_ref( |
| 3002 |
&self, |
| 3003 |
k: &str, |
| 3004 |
) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> { |
| 3005 |
self.0.read_ref(k) |
| 3006 |
} |
| 3007 |
fn write_ref( |
| 3008 |
&self, |
| 3009 |
k: &str, |
| 3010 |
b: &[u8], |
| 3011 |
e: Option<&crate::strata::Version>, |
| 3012 |
) -> crate::strata::Result<crate::strata::Version> { |
| 3013 |
self.0.write_ref(k, b, e) |
| 3014 |
} |
| 3015 |
} |
| 3016 |
let dir = tempfile::tempdir().unwrap(); |
| 3017 |
let backend = Arc::new(Counting(MemoryStore::new(), AtomicUsize::new(0))); |
| 3018 |
let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap(); |
| 3019 |
let refs: Vec<ObjectRef> = (0..10) |
| 3020 |
.map(|i| { |
| 3021 |
s.put(ObjectType::Chunk, format!("p {i}").as_bytes()) |
| 3022 |
.unwrap() |
| 3023 |
}) |
| 3024 |
.collect(); |
| 3025 |
s.pack_loose().unwrap(); |
| 3026 |
let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap(); |
| 3027 |
backend.1.store(0, Ordering::Relaxed); |
| 3028 |
for r in &refs { |
| 3029 |
assert!(s.get(r).is_ok()); |
| 3030 |
} |
| 3031 |
assert!(s.get_many(&refs).iter().all(|g| g.is_ok())); |
| 3032 |
assert_eq!(backend.1.load(Ordering::Relaxed), 0, "no loose lookups"); |
| 3033 |
|
| 3034 |
let missing = hash_object(ObjectType::Chunk, b"never stored"); |
| 3035 |
assert!(matches!(s.get(&missing), Err(StoreError::NotFound(_)))); |
| 3036 |
assert!(backend.1.load(Ordering::Relaxed) > 0); |
| 3037 |
} |
| 3038 |
} |
| 3039 |
|
| 3040 |
|
| 3041 |
|
| 3042 |
#[cfg(test)] |
| 3043 |
pub(crate) mod reader_tests { |
| 3044 |
use super::*; |
| 3045 |
|
| 3046 |
fn chunk(i: usize) -> Vec<u8> { |
| 3047 |
format!("object {i} ").repeat(1 + i % 5).into_bytes() |
| 3048 |
} |
| 3049 |
|
| 3050 |
|
| 3051 |
fn each_pair(f: impl Fn(&str, Store, Store)) { |
| 3052 |
let dir = tempfile::tempdir().unwrap(); |
| 3053 |
let s = Store::open(dir.path().join("objects")).unwrap(); |
| 3054 |
let r = Store::open(dir.path().join("objects")).unwrap(); |
| 3055 |
f("directory", s, r); |
| 3056 |
let m: Arc<Backend> = Arc::new(MemoryStore::new()); |
| 3057 |
let s = Store::open_on(m.clone(), dir.path().join("scratch")).unwrap(); |
| 3058 |
let r = Store::open_on(m, dir.path().join("scratch-r")).unwrap(); |
| 3059 |
f("memory", s, r); |
| 3060 |
} |
| 3061 |
|
| 3062 |
|
| 3063 |
pub(crate) fn age_retired(s: &Store, by: std::time::Duration) { |
| 3064 |
s.update_retired(|m| { |
| 3065 |
for t in m.values_mut() { |
| 3066 |
*t = t.saturating_sub(by.as_millis() as u64); |
| 3067 |
} |
| 3068 |
}) |
| 3069 |
.unwrap(); |
| 3070 |
} |
| 3071 |
|
| 3072 |
fn retired_keys(s: &Store) -> Vec<String> { |
| 3073 |
s.read_retired().unwrap().0.into_keys().collect() |
| 3074 |
} |
| 3075 |
|
| 3076 |
#[test] |
| 3077 |
fn a_reader_opened_before_a_repack_reads_every_object_after_it() { |
| 3078 |
each_pair(|which, writer, _| { |
| 3079 |
let mut refs = Vec::new(); |
| 3080 |
for i in 0..40 { |
| 3081 |
refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap()); |
| 3082 |
} |
| 3083 |
writer.pack_loose().unwrap(); |
| 3084 |
for i in 40..80 { |
| 3085 |
refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap()); |
| 3086 |
} |
| 3087 |
|
| 3088 |
let reader = Store::open_on(writer.backend().clone(), writer.dir()).unwrap(); |
| 3089 |
assert_eq!(reader.pack_count(), 1, "{which}"); |
| 3090 |
let (_, merged) = writer.repack_all(REPACK_ZSTD_LEVEL).unwrap(); |
| 3091 |
assert_eq!(merged, 1); |
| 3092 |
|
| 3093 |
let retired = retired_keys(&writer); |
| 3094 |
assert_eq!(retired.len(), 81, "{which}: {retired:?}"); |
| 3095 |
assert_eq!(writer.loose_count().unwrap(), 0, "retired is not loose"); |
| 3096 |
|
| 3097 |
for (i, r) in refs.iter().enumerate() { |
| 3098 |
assert_eq!(reader.get(r).unwrap().1, chunk(i), "{which}: {i}"); |
| 3099 |
} |
| 3100 |
assert_eq!(reader.pack_count(), 1); |
| 3101 |
|
| 3102 |
let fresh = Store::open_on(writer.backend().clone(), writer.dir()).unwrap(); |
| 3103 |
assert_eq!(fresh.pack_count(), 1); |
| 3104 |
assert_ne!(fresh.packs.borrow()[0].key, reader.packs.borrow()[0].key); |
| 3105 |
|
| 3106 |
|
| 3107 |
let report = writer |
| 3108 |
.expire_retired(std::time::Duration::ZERO, &mut silent()) |
| 3109 |
.unwrap(); |
| 3110 |
assert_eq!((report.packs, report.loose), (1, 80), "{which}"); |
| 3111 |
assert!(retired_keys(&writer).is_empty()); |
| 3112 |
for (i, r) in refs.iter().enumerate() { |
| 3113 |
assert_eq!(reader.get(r).unwrap().1, chunk(i), "{which}: {i}"); |
| 3114 |
} |
| 3115 |
let other = Store::open_on(writer.backend().clone(), writer.dir()).unwrap(); |
| 3116 |
let before = other.packs.borrow()[0].key.clone(); |
| 3117 |
writer.put(ObjectType::Chunk, b"new").unwrap(); |
| 3118 |
writer.repack_all(REPACK_ZSTD_LEVEL).unwrap(); |
| 3119 |
writer |
| 3120 |
.expire_retired(std::time::Duration::ZERO, &mut silent()) |
| 3121 |
.unwrap(); |
| 3122 |
assert!(other.get_many(&refs).iter().all(|g| g.is_ok()), "{which}"); |
| 3123 |
assert_ne!(other.packs.borrow()[0].key, before); |
| 3124 |
let third = Store::open_on(writer.backend().clone(), writer.dir()).unwrap(); |
| 3125 |
writer.put(ObjectType::Chunk, b"newer").unwrap(); |
| 3126 |
writer.repack_all(REPACK_ZSTD_LEVEL).unwrap(); |
| 3127 |
writer |
| 3128 |
.expire_retired(std::time::Duration::ZERO, &mut silent()) |
| 3129 |
.unwrap(); |
| 3130 |
assert!(refs.iter().all(|r| third.has(r)), "{which}"); |
| 3131 |
assert!(!third.has(&hash_object(ObjectType::Chunk, b"never"))); |
| 3132 |
}); |
| 3133 |
} |
| 3134 |
|
| 3135 |
#[test] |
| 3136 |
fn a_reader_opened_before_a_spilled_batch_reads_its_objects() { |
| 3137 |
each_pair(|which, writer, reader| { |
| 3138 |
assert_eq!(reader.pack_count(), 0); |
| 3139 |
let mut refs = Vec::new(); |
| 3140 |
{ |
| 3141 |
let batch = writer.begin_batch().unwrap(); |
| 3142 |
for i in 0..(SPILL_OBJECTS + 50) { |
| 3143 |
refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap()); |
| 3144 |
} |
| 3145 |
batch.finish().unwrap(); |
| 3146 |
} |
| 3147 |
assert_eq!(writer.pack_count(), 1, "{which}: the batch spilled"); |
| 3148 |
assert_eq!(reader.get(&refs[0]).unwrap().1, chunk(0), "{which}"); |
| 3149 |
assert_eq!(reader.pack_count(), 1, "the miss loaded the new pack"); |
| 3150 |
let got = reader.get_many(&refs); |
| 3151 |
assert!(got.iter().all(|g| g.is_ok()), "{which}"); |
| 3152 |
|
| 3153 |
let late = Store::open_on(writer.backend().clone(), writer.dir()).unwrap(); |
| 3154 |
{ |
| 3155 |
let batch = writer.begin_batch().unwrap(); |
| 3156 |
for i in 1000..(1000 + SPILL_OBJECTS + 1) { |
| 3157 |
writer.put(ObjectType::Chunk, &chunk(i)).unwrap(); |
| 3158 |
} |
| 3159 |
batch.finish().unwrap(); |
| 3160 |
} |
| 3161 |
assert!(late.has(&hash_object(ObjectType::Chunk, &chunk(1000)))); |
| 3162 |
}); |
| 3163 |
} |
| 3164 |
|
| 3165 |
#[test] |
| 3166 |
fn retired_entries_expire_after_the_hour_and_are_never_loaded() { |
| 3167 |
each_pair(|which, s, _| { |
| 3168 |
for i in 0..10 { |
| 3169 |
s.put(ObjectType::Chunk, &chunk(i)).unwrap(); |
| 3170 |
} |
| 3171 |
s.pack_loose().unwrap(); |
| 3172 |
s.put(ObjectType::Chunk, &chunk(10)).unwrap(); |
| 3173 |
s.repack_all(ZSTD_LEVEL).unwrap(); |
| 3174 |
|
| 3175 |
age_retired(&s, RETIRED_FOR + std::time::Duration::from_secs(60)); |
| 3176 |
let old = retired_keys(&s); |
| 3177 |
assert_eq!(old.len(), 12, "{which}: {old:?}"); |
| 3178 |
|
| 3179 |
s.put(ObjectType::Chunk, &chunk(11)).unwrap(); |
| 3180 |
s.repack_all(ZSTD_LEVEL).unwrap(); |
| 3181 |
assert_eq!(retired_keys(&s).len(), 14); |
| 3182 |
let listed = s |
| 3183 |
.backend() |
| 3184 |
.list("pack-") |
| 3185 |
.unwrap() |
| 3186 |
.into_iter() |
| 3187 |
.filter(|e| e.key.ends_with(".idx")) |
| 3188 |
.count(); |
| 3189 |
assert_eq!(listed, 3, "{which}"); |
| 3190 |
let fresh = Store::open_on(s.backend().clone(), s.dir()).unwrap(); |
| 3191 |
assert_eq!( |
| 3192 |
fresh.pack_count(), |
| 3193 |
1, |
| 3194 |
"{which}: retired packs are not loaded" |
| 3195 |
); |
| 3196 |
let report = s.expire_retired(RETIRED_FOR, &mut silent()).unwrap(); |
| 3197 |
assert_eq!( |
| 3198 |
report, |
| 3199 |
RetiredReport { |
| 3200 |
packs: 1, |
| 3201 |
loose: 11, |
| 3202 |
kept: 2 |
| 3203 |
}, |
| 3204 |
"{which}" |
| 3205 |
); |
| 3206 |
let young = retired_keys(&s); |
| 3207 |
assert_eq!(young.len(), 2, "{which}: {young:?}"); |
| 3208 |
assert!(young.iter().all(|k| !old.contains(k))); |
| 3209 |
for k in old.iter().filter(|k| k.ends_with(".pack")) { |
| 3210 |
assert!(!s.backend().has(k).unwrap(), "{which}: {k}"); |
| 3211 |
} |
| 3212 |
for k in &young { |
| 3213 |
assert!(s.backend().has(k).unwrap(), "{which}: {k} kept"); |
| 3214 |
} |
| 3215 |
for i in 0..12 { |
| 3216 |
let r = hash_object(ObjectType::Chunk, &chunk(i)); |
| 3217 |
assert_eq!(fresh.get(&r).unwrap().1, chunk(i), "{which}"); |
| 3218 |
} |
| 3219 |
}); |
| 3220 |
} |
| 3221 |
|
| 3222 |
|
| 3223 |
|
| 3224 |
#[test] |
| 3225 |
fn a_retired_loose_object_no_pack_holds_is_kept() { |
| 3226 |
each_pair(|which, s, _| { |
| 3227 |
let a = s.put(ObjectType::Chunk, b"a").unwrap(); |
| 3228 |
let b = s.put(ObjectType::Chunk, b"b").unwrap(); |
| 3229 |
s.pack_loose().unwrap(); |
| 3230 |
assert_eq!(s.loose_count().unwrap(), 0); |
| 3231 |
|
| 3232 |
let first = s.packs.borrow()[0].key.clone(); |
| 3233 |
s.write_pack(&[a], &[], &[first], ZSTD_LEVEL, &mut silent()) |
| 3234 |
.unwrap(); |
| 3235 |
let report = s |
| 3236 |
.expire_retired(std::time::Duration::ZERO, &mut silent()) |
| 3237 |
.unwrap(); |
| 3238 |
assert_eq!((report.packs, report.loose), (1, 1), "{which}: a's copy"); |
| 3239 |
assert!(retired_keys(&s).is_empty()); |
| 3240 |
assert_eq!(s.get(&b).unwrap().1, b"b", "{which}"); |
| 3241 |
assert_eq!(s.get(&a).unwrap().1, b"a"); |
| 3242 |
assert_eq!(s.loose_count().unwrap(), 1, "b is loose again"); |
| 3243 |
|
| 3244 |
let keep = [a].into_iter().collect(); |
| 3245 |
let r = s.prune(&keep, std::time::Duration::ZERO).unwrap(); |
| 3246 |
assert_eq!(r.objects, 1); |
| 3247 |
assert!(!s.has(&b)); |
| 3248 |
}); |
| 3249 |
} |
| 3250 |
|
| 3251 |
|
| 3252 |
|
| 3253 |
pub(crate) struct Racing { |
| 3254 |
inner: MemoryStore, |
| 3255 |
lists: std::sync::atomic::AtomicUsize, |
| 3256 |
|
| 3257 |
pub(crate) doomed: std::sync::Mutex<Option<String>>, |
| 3258 |
|
| 3259 |
ghost: bool, |
| 3260 |
} |
| 3261 |
|
| 3262 |
impl crate::strata::Store for Racing { |
| 3263 |
fn describe(&self) -> String { |
| 3264 |
"racing".into() |
| 3265 |
} |
| 3266 |
fn has(&self, k: &str) -> crate::strata::Result<bool> { |
| 3267 |
self.inner.has(k) |
| 3268 |
} |
| 3269 |
fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> { |
| 3270 |
self.inner.get(k) |
| 3271 |
} |
| 3272 |
fn get_range(&self, k: &str, r: std::ops::Range<u64>) -> crate::strata::Result<Vec<u8>> { |
| 3273 |
self.inner.get_range(k, r) |
| 3274 |
} |
| 3275 |
fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> { |
| 3276 |
self.inner.put_new(k, b) |
| 3277 |
} |
| 3278 |
fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> { |
| 3279 |
let mut out = self.inner.list(p)?; |
| 3280 |
if p == "pack-" { |
| 3281 |
self.lists |
| 3282 |
.fetch_add(1, std::sync::atomic::Ordering::Relaxed); |
| 3283 |
if let Some(k) = self.doomed.lock().unwrap().take() { |
| 3284 |
self.inner.delete(&k)?; |
| 3285 |
} |
| 3286 |
if self.ghost { |
| 3287 |
for k in ["pack-0000.idx", "pack-0000.pack"] { |
| 3288 |
out.push(crate::strata::Entry { |
| 3289 |
key: k.into(), |
| 3290 |
size: 100, |
| 3291 |
modified_ms: None, |
| 3292 |
}); |
| 3293 |
} |
| 3294 |
out.sort_by(|a, b| a.key.cmp(&b.key)); |
| 3295 |
} |
| 3296 |
} |
| 3297 |
Ok(out) |
| 3298 |
} |
| 3299 |
fn delete(&self, k: &str) -> crate::strata::Result<()> { |
| 3300 |
self.inner.delete(k) |
| 3301 |
} |
| 3302 |
fn read_ref( |
| 3303 |
&self, |
| 3304 |
k: &str, |
| 3305 |
) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> { |
| 3306 |
self.inner.read_ref(k) |
| 3307 |
} |
| 3308 |
fn write_ref( |
| 3309 |
&self, |
| 3310 |
k: &str, |
| 3311 |
b: &[u8], |
| 3312 |
e: Option<&crate::strata::Version>, |
| 3313 |
) -> crate::strata::Result<crate::strata::Version> { |
| 3314 |
self.inner.write_ref(k, b, e) |
| 3315 |
} |
| 3316 |
} |
| 3317 |
|
| 3318 |
pub(crate) fn racing(ghost: bool) -> Arc<Racing> { |
| 3319 |
Arc::new(Racing { |
| 3320 |
inner: MemoryStore::new(), |
| 3321 |
lists: Default::default(), |
| 3322 |
doomed: Default::default(), |
| 3323 |
ghost, |
| 3324 |
}) |
| 3325 |
} |
| 3326 |
|
| 3327 |
#[test] |
| 3328 |
fn an_index_deleted_between_listing_and_loading_is_passed_over() { |
| 3329 |
let dir = tempfile::tempdir().unwrap(); |
| 3330 |
let backend = racing(false); |
| 3331 |
let s = Store::open_on(backend.clone(), dir.path()).unwrap(); |
| 3332 |
let a = s.put(ObjectType::Chunk, b"a").unwrap(); |
| 3333 |
s.pack_loose().unwrap(); |
| 3334 |
let b = s.put(ObjectType::Chunk, b"b").unwrap(); |
| 3335 |
s.pack_loose().unwrap(); |
| 3336 |
let idx = { |
| 3337 |
let packs = s.packs.borrow(); |
| 3338 |
let (i, _) = s.find_packed(&a).unwrap(); |
| 3339 |
format!("{}.idx", packs[i].key.trim_end_matches(".pack")) |
| 3340 |
}; |
| 3341 |
*backend.doomed.lock().unwrap() = Some(idx); |
| 3342 |
let before = backend.lists.load(std::sync::atomic::Ordering::Relaxed); |
| 3343 |
let again = Store::open_on(backend.clone(), dir.path()).unwrap(); |
| 3344 |
assert_eq!(again.pack_count(), 1, "the deleted one is passed over"); |
| 3345 |
assert_eq!( |
| 3346 |
backend.lists.load(std::sync::atomic::Ordering::Relaxed) - before, |
| 3347 |
2, |
| 3348 |
"and the store listed again" |
| 3349 |
); |
| 3350 |
assert_eq!(again.get(&b).unwrap().1, b"b"); |
| 3351 |
|
| 3352 |
let ghosted = racing(true); |
| 3353 |
let s = Store::open_on(ghosted.clone(), dir.path().join("g")).unwrap(); |
| 3354 |
let c = s.put(ObjectType::Chunk, b"c").unwrap(); |
| 3355 |
s.pack_loose().unwrap(); |
| 3356 |
let again = Store::open_on(ghosted, dir.path().join("g")).unwrap(); |
| 3357 |
assert_eq!(again.pack_count(), 1); |
| 3358 |
assert_eq!(again.get(&c).unwrap().1, b"c"); |
| 3359 |
} |
| 3360 |
} |
| 3361 |
|
| 3362 |
#[cfg(test)] |
| 3363 |
mod batch_tests { |
| 3364 |
use super::*; |
| 3365 |
|
| 3366 |
#[test] |
| 3367 |
fn a_large_batch_packs_every_object_and_indexes_them_all() { |
| 3368 |
let dir = tempfile::tempdir().unwrap(); |
| 3369 |
let store = Store::open(dir.path()).unwrap(); |
| 3370 |
let mut refs = Vec::new(); |
| 3371 |
{ |
| 3372 |
let batch = store.begin_batch().unwrap(); |
| 3373 |
for i in 0..1000u32 { |
| 3374 |
let body = format!("object {i} {}", "x".repeat(20_000 + i as usize)).into_bytes(); |
| 3375 |
refs.push(store.put(ObjectType::Blob, &body).unwrap()); |
| 3376 |
} |
| 3377 |
batch.finish().unwrap(); |
| 3378 |
} |
| 3379 |
let again = Store::open(dir.path()).unwrap(); |
| 3380 |
for r in &refs { |
| 3381 |
again |
| 3382 |
.get(r) |
| 3383 |
.unwrap_or_else(|e| panic!("{}: {e}", r.short(12))); |
| 3384 |
} |
| 3385 |
} |
| 3386 |
|
| 3387 |
#[test] |
| 3388 |
fn a_pack_a_stopped_batch_left_is_adopted_and_its_torn_end_cut() { |
| 3389 |
|
| 3390 |
|
| 3391 |
let dir = tempfile::tempdir().unwrap(); |
| 3392 |
let mut refs = Vec::new(); |
| 3393 |
{ |
| 3394 |
let store = Store::open(dir.path()).unwrap(); |
| 3395 |
let batch = store.begin_batch().unwrap(); |
| 3396 |
for i in 0..300u32 { |
| 3397 |
let body = format!("object {i} {}", "y".repeat(20_000 + i as usize)).into_bytes(); |
| 3398 |
refs.push(store.put(ObjectType::Blob, &body).unwrap()); |
| 3399 |
} |
| 3400 |
std::mem::forget(batch); |
| 3401 |
} |
| 3402 |
let tmp: Vec<_> = fs::read_dir(dir.path()) |
| 3403 |
.unwrap() |
| 3404 |
.flatten() |
| 3405 |
.filter(|e| e.file_name().to_string_lossy().starts_with(".tmp-")) |
| 3406 |
.map(|e| e.path()) |
| 3407 |
.collect(); |
| 3408 |
assert_eq!(tmp.len(), 1, "one unfinished pack"); |
| 3409 |
|
| 3410 |
let mut f = fs::OpenOptions::new().append(true).open(&tmp[0]).unwrap(); |
| 3411 |
f.write_all(&[0, 1, 0xff, 0, 0, 0, 9, 9, 0, 0]).unwrap(); |
| 3412 |
f.write_all(&[7; 40]).unwrap(); |
| 3413 |
drop(f); |
| 3414 |
let store = Store::open(dir.path()).unwrap(); |
| 3415 |
assert!(store.get(&refs[0]).is_err(), "unreadable before"); |
| 3416 |
let young = store |
| 3417 |
.adopt_unfinished_packs(std::time::Duration::from_secs(3600), &|_| {}) |
| 3418 |
.unwrap(); |
| 3419 |
assert_eq!(young, (0, 0), "a pack written to lately is left alone"); |
| 3420 |
let (packs, objects) = store |
| 3421 |
.adopt_unfinished_packs(std::time::Duration::ZERO, &|_| {}) |
| 3422 |
.unwrap(); |
| 3423 |
assert_eq!((packs, objects), (1, 300)); |
| 3424 |
for r in &refs { |
| 3425 |
store |
| 3426 |
.get(r) |
| 3427 |
.unwrap_or_else(|e| panic!("{}: {e}", r.short(12))); |
| 3428 |
} |
| 3429 |
let again = Store::open(dir.path()).unwrap(); |
| 3430 |
assert!(again.get(&refs[299]).is_ok(), "indexed for good"); |
| 3431 |
} |
| 3432 |
} |
| 3433 |
|
| 3434 |
#[cfg(test)] |
| 3435 |
mod size_tests { |
| 3436 |
use super::*; |
| 3437 |
|
| 3438 |
#[test] |
| 3439 |
fn an_object_over_the_limit_is_refused_and_nothing_is_written() { |
| 3440 |
let dir = tempfile::tempdir().unwrap(); |
| 3441 |
let store = Store::open(dir.path()).unwrap(); |
| 3442 |
let big = vec![b'x'; MAX_OBJECT_LEN as usize + 1]; |
| 3443 |
{ |
| 3444 |
let batch = store.begin_batch().unwrap(); |
| 3445 |
store.put(ObjectType::Blob, b"small").unwrap(); |
| 3446 |
let e = store.put(ObjectType::Change, &big).unwrap_err(); |
| 3447 |
assert!( |
| 3448 |
matches!(e, StoreError::TooLarge { kind: "change", .. }), |
| 3449 |
"{e}" |
| 3450 |
); |
| 3451 |
assert!(e.to_string().contains("commit it in parts"), "{e}"); |
| 3452 |
batch.finish().unwrap(); |
| 3453 |
} |
| 3454 |
|
| 3455 |
let again = Store::open(dir.path()).unwrap(); |
| 3456 |
again.get(&hash_object(ObjectType::Blob, b"small")).unwrap(); |
| 3457 |
} |
| 3458 |
} |
| 3459 |
|
| 3460 |
#[cfg(test)] |
| 3461 |
mod text_tests { |
| 3462 |
use super::*; |
| 3463 |
use crate::objects::{AtomRef, Edit, CHANGE_VERSION}; |
| 3464 |
|
| 3465 |
fn large_change() -> Change { |
| 3466 |
let mut c = crate::objects::tests::sample_change(); |
| 3467 |
assert_eq!(c.version, CHANGE_VERSION); |
| 3468 |
c.edits[0] = Edit::Insert { |
| 3469 |
node: crate::ids::NodeId([7; 16]), |
| 3470 |
after: AtomRef::Start, |
| 3471 |
before: AtomRef::End, |
| 3472 |
lines: (0..2000) |
| 3473 |
.map(|i| format!("line {i}\n").into_bytes()) |
| 3474 |
.collect(), |
| 3475 |
}; |
| 3476 |
c |
| 3477 |
} |
| 3478 |
|
| 3479 |
#[test] |
| 3480 |
fn a_large_insert_is_stored_beside_its_change_and_read_back() { |
| 3481 |
let dir = tempfile::tempdir().unwrap(); |
| 3482 |
let s = Store::open(dir.path()).unwrap(); |
| 3483 |
let c = large_change(); |
| 3484 |
let r = s.put_change(&c).unwrap(); |
| 3485 |
assert_eq!(r, c.hash()); |
| 3486 |
let (_, body) = s.get(&r).unwrap(); |
| 3487 |
assert!( |
| 3488 |
body.len() < 1024, |
| 3489 |
"the change holds a reference, not {} bytes of lines", |
| 3490 |
body.len() |
| 3491 |
); |
| 3492 |
let (back, parts) = s.get_change_parts(&r).unwrap(); |
| 3493 |
assert_eq!(back, c); |
| 3494 |
|
| 3495 |
assert_eq!(parts.len(), 2); |
| 3496 |
for p in &parts { |
| 3497 |
assert!(s.has(p)); |
| 3498 |
} |
| 3499 |
|
| 3500 |
let n = s.list().unwrap().len(); |
| 3501 |
s.put_change(&c).unwrap(); |
| 3502 |
assert_eq!(s.list().unwrap().len(), n); |
| 3503 |
} |
| 3504 |
|
| 3505 |
#[test] |
| 3506 |
fn a_change_whose_text_blob_is_missing_is_reported_not_misread() { |
| 3507 |
if !super::files_on_disk() { |
| 3508 |
return; |
| 3509 |
} |
| 3510 |
let dir = tempfile::tempdir().unwrap(); |
| 3511 |
let s = Store::open(dir.path()).unwrap(); |
| 3512 |
let c = large_change(); |
| 3513 |
let r = s.put_change(&c).unwrap(); |
| 3514 |
let (_, parts) = s.get_change_parts(&r).unwrap(); |
| 3515 |
fs::remove_file(s.path_for(&parts[0])).unwrap(); |
| 3516 |
let e = s.get_change(&r).unwrap_err(); |
| 3517 |
assert!( |
| 3518 |
matches!(&e, StoreError::Corrupt(h, why) if *h == r && why.contains("missing")), |
| 3519 |
"{e}" |
| 3520 |
); |
| 3521 |
} |
| 3522 |
} |
| 3523 |
|
| 3524 |
#[cfg(test)] |
| 3525 |
mod put_many_tests { |
| 3526 |
use super::*; |
| 3527 |
|
| 3528 |
#[test] |
| 3529 |
fn many_objects_in_a_batch_are_compressed_together_and_all_read_back() { |
| 3530 |
let dir = tempfile::tempdir().unwrap(); |
| 3531 |
let s = Store::open(dir.path()).unwrap(); |
| 3532 |
let before = s.put(ObjectType::Chunk, b"already here").unwrap(); |
| 3533 |
let bodies: Vec<Vec<u8>> = (0..200u32) |
| 3534 |
.map(|i| format!("chunk {i} ").repeat(6000).into_bytes()) |
| 3535 |
.collect(); |
| 3536 |
let mut objects: Vec<(ObjectType, &[u8])> = bodies |
| 3537 |
.iter() |
| 3538 |
.map(|b| (ObjectType::Chunk, b.as_slice())) |
| 3539 |
.collect(); |
| 3540 |
objects.push((ObjectType::Chunk, bodies[0].as_slice())); |
| 3541 |
objects.push((ObjectType::Chunk, b"already here")); |
| 3542 |
let refs = { |
| 3543 |
let batch = s.begin_batch().unwrap(); |
| 3544 |
let refs = s.put_many(&objects).unwrap(); |
| 3545 |
batch.finish().unwrap(); |
| 3546 |
refs |
| 3547 |
}; |
| 3548 |
assert_eq!(refs.len(), objects.len()); |
| 3549 |
assert_eq!(refs[200], refs[0], "a repeat is the same object"); |
| 3550 |
assert_eq!(refs[201], before); |
| 3551 |
let again = Store::open(dir.path()).unwrap(); |
| 3552 |
for (r, b) in refs.iter().zip(&objects) { |
| 3553 |
assert_eq!(again.get(r).unwrap().1, b.1); |
| 3554 |
} |
| 3555 |
} |
| 3556 |
} |