|
1 |
+ |
//! Oxygen Strata (docs/stores.md): the storage a repository can be kept |
|
2 |
+ |
//! on, as a trait, so that a bucket, a directory or a hub's disk all look |
|
3 |
+ |
//! the same to the code that pushes to and fetches from them. |
|
4 |
+ |
//! |
|
5 |
+ |
//! A store has two halves. **Objects** are written once and never |
|
6 |
+ |
//! change: packs and their indexes, named by content, which any store |
|
7 |
+ |
//! can hold safely. **References** are the one mutable record (bookmarks, |
|
8 |
+ |
//! tags, the policy), updated only by compare-and-swap on a version the |
|
9 |
+ |
//! store hands out: an ETag on S3, a generation on GCS, a content hash on |
|
10 |
+ |
//! a directory. Two writers can therefore share a store with no lock |
|
11 |
+ |
//! server: the loser of a race reads again, merges, and tries again. |
|
12 |
+ |
//! |
|
13 |
+ |
//! The working repository (`.oxy/`) is not a `Store`; it is read and |
|
14 |
+ |
//! written directly, as before. A store is what a *remote* is kept on. |
|
15 |
+ |
//! This module holds the trait, a directory store, an in-memory store |
|
16 |
+ |
//! for tests, and the conformance check every store must pass. Cloud |
|
17 |
+ |
//! stores live in a crate of their own, because of what they pull in. |
|
18 |
+ |
|
|
19 |
+ |
use std::collections::BTreeMap; |
|
20 |
+ |
use std::fs; |
|
21 |
+ |
use std::io::{self, Read, Seek, SeekFrom, Write}; |
|
22 |
+ |
use std::ops::Range; |
|
23 |
+ |
use std::path::{Path, PathBuf}; |
|
24 |
+ |
use std::sync::Mutex; |
|
25 |
+ |
use std::time::{Duration, Instant}; |
|
26 |
+ |
|
|
27 |
+ |
/// The version of a reference record as the store reports it: opaque, |
|
28 |
+ |
/// compared only for equality, and only ever valid for the store that |
|
29 |
+ |
/// issued it. Writing the same bytes again may keep the version (S3's |
|
30 |
+ |
/// ETag and a directory's content hash are both of the content): a |
|
31 |
+ |
/// compare-and-swap guards against losing a *different* record, which is |
|
32 |
+ |
/// all it needs to. |
|
33 |
+ |
#[derive(Clone, Debug, PartialEq, Eq)] |
|
34 |
+ |
pub struct Version(pub String); |
|
35 |
+ |
|
|
36 |
+ |
/// One key in a listing. |
|
37 |
+ |
#[derive(Clone, Debug, PartialEq, Eq)] |
|
38 |
+ |
pub struct Entry { |
|
39 |
+ |
pub key: String, |
|
40 |
+ |
pub size: u64, |
|
41 |
+ |
} |
|
42 |
+ |
|
|
43 |
+ |
#[derive(Debug, thiserror::Error)] |
|
44 |
+ |
pub enum Error { |
|
45 |
+ |
#[error("{0}: not in the store")] |
|
46 |
+ |
NotFound(String), |
|
47 |
+ |
#[error("{0}: already in the store")] |
|
48 |
+ |
Exists(String), |
|
49 |
+ |
#[error("{0}: changed since it was read")] |
|
50 |
+ |
Conflict(String), |
|
51 |
+ |
#[error("bad store key {0:?}: {1}")] |
|
52 |
+ |
Key(String, &'static str), |
|
53 |
+ |
#[error("{0}: busy (another writer holds it)")] |
|
54 |
+ |
Busy(String), |
|
55 |
+ |
#[error(transparent)] |
|
56 |
+ |
Io(#[from] io::Error), |
|
57 |
+ |
#[error("{0}")] |
|
58 |
+ |
Backend(String), |
|
59 |
+ |
} |
|
60 |
+ |
|
|
61 |
+ |
pub type Result<T> = std::result::Result<T, Error>; |
|
62 |
+ |
|
|
63 |
+ |
/// Longest key a store accepts; S3 allows 1024 bytes. |
|
64 |
+ |
pub const MAX_KEY_LEN: usize = 1024; |
|
65 |
+ |
|
|
66 |
+ |
/// Refuse a key that is not a plain relative path: `/`-separated |
|
67 |
+ |
/// segments of `[A-Za-z0-9._-]`, none empty, none `.` or `..`. The same |
|
68 |
+ |
/// rule on every store, so a key written to one can be read from another, |
|
69 |
+ |
/// and so a key can never escape a directory store's root. |
|
70 |
+ |
pub fn check_key(key: &str) -> Result<()> { |
|
71 |
+ |
let bad = |why| Error::Key(key.to_string(), why); |
|
72 |
+ |
if key.is_empty() { |
|
73 |
+ |
return Err(bad("empty")); |
|
74 |
+ |
} |
|
75 |
+ |
if key.len() > MAX_KEY_LEN { |
|
76 |
+ |
return Err(bad("longer than 1024 bytes")); |
|
77 |
+ |
} |
|
78 |
+ |
for seg in key.split('/') { |
|
79 |
+ |
if seg.is_empty() { |
|
80 |
+ |
return Err(bad( |
|
81 |
+ |
"an empty segment (a leading, trailing or doubled slash)", |
|
82 |
+ |
)); |
|
83 |
+ |
} |
|
84 |
+ |
if seg == "." || seg == ".." { |
|
85 |
+ |
return Err(bad("a . or .. segment")); |
|
86 |
+ |
} |
|
87 |
+ |
if !seg |
|
88 |
+ |
.bytes() |
|
89 |
+ |
.all(|b| b.is_ascii_alphanumeric() || b == b'.' || b == b'_' || b == b'-') |
|
90 |
+ |
{ |
|
91 |
+ |
return Err(bad("a character outside A-Z a-z 0-9 . _ -")); |
|
92 |
+ |
} |
|
93 |
+ |
} |
|
94 |
+ |
Ok(()) |
|
95 |
+ |
} |
|
96 |
+ |
|
|
97 |
+ |
/// Where a repository's packs and references are kept. Every method is |
|
98 |
+ |
/// synchronous and may be called from several threads at once. |
|
99 |
+ |
pub trait Store: Send + Sync { |
|
100 |
+ |
/// A name for messages: the URL, or the directory. |
|
101 |
+ |
fn describe(&self) -> String; |
|
102 |
+ |
|
|
103 |
+ |
// ---- objects: written once, never changed |
|
104 |
+ |
|
|
105 |
+ |
fn has(&self, key: &str) -> Result<bool>; |
|
106 |
+ |
/// The whole object; `NotFound` when there is none. |
|
107 |
+ |
fn get(&self, key: &str) -> Result<Vec<u8>>; |
|
108 |
+ |
/// Bytes `range` of an object; an error when the range runs past its end. |
|
109 |
+ |
fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>>; |
|
110 |
+ |
/// Write an object that does not exist yet; `Exists` when it does, and |
|
111 |
+ |
/// then nothing is changed. Readers never see a partial object. |
|
112 |
+ |
fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()>; |
|
113 |
+ |
/// Every key under `prefix` (a key prefix, not necessarily a whole |
|
114 |
+ |
/// segment), sorted, with sizes. |
|
115 |
+ |
fn list(&self, prefix: &str) -> Result<Vec<Entry>>; |
|
116 |
+ |
/// Remove an object; fine when it is already gone. |
|
117 |
+ |
fn delete(&self, key: &str) -> Result<()>; |
|
118 |
+ |
|
|
119 |
+ |
// ---- references: one small record, updated by compare-and-swap |
|
120 |
+ |
|
|
121 |
+ |
/// The record and its current version, or `None` when it was never |
|
122 |
+ |
/// written. |
|
123 |
+ |
fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>>; |
|
124 |
+ |
/// Replace the record only if its version is still `expected` |
|
125 |
+ |
/// (`None`: only if it does not exist yet). `Conflict` otherwise, and |
|
126 |
+ |
/// then nothing is changed. Returns the new version. |
|
127 |
+ |
fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version>; |
|
128 |
+ |
} |
|
129 |
+ |
|
|
130 |
+ |
/// A reference record's version on stores that have no native one: the |
|
131 |
+ |
/// hash of its bytes, which two different records never share. |
|
132 |
+ |
fn content_version(bytes: &[u8]) -> Version { |
|
133 |
+ |
Version(blake3::hash(bytes).to_hex().to_string()) |
|
134 |
+ |
} |
|
135 |
+ |
|
|
136 |
+ |
fn check_expected(key: &str, current: Option<&Version>, expected: Option<&Version>) -> Result<()> { |
|
137 |
+ |
if current == expected { |
|
138 |
+ |
Ok(()) |
|
139 |
+ |
} else { |
|
140 |
+ |
Err(Error::Conflict(key.to_string())) |
|
141 |
+ |
} |
|
142 |
+ |
} |
|
143 |
+ |
|
|
144 |
+ |
// ---------------------------------------------------------------- directory |
|
145 |
+ |
|
|
146 |
+ |
/// A store in a directory: each key a file under the root. Create-only |
|
147 |
+ |
/// writes go through a hard link, which the file system refuses when the |
|
148 |
+ |
/// name is taken, and a reference is updated under a lock file beside it. |
|
149 |
+ |
/// Several processes on one machine can share it; it is also what the |
|
150 |
+ |
/// tests and `oxy remote check` use. |
|
151 |
+ |
pub struct LocalStore { |
|
152 |
+ |
root: PathBuf, |
|
153 |
+ |
} |
|
154 |
+ |
|
|
155 |
+ |
/// How long a reference's lock may be held before it counts as abandoned |
|
156 |
+ |
/// by a process that died, and how long a writer waits for it. |
|
157 |
+ |
const LOCK_STALE: Duration = Duration::from_secs(60); |
|
158 |
+ |
const LOCK_WAIT: Duration = Duration::from_secs(10); |
|
159 |
+ |
|
|
160 |
+ |
impl LocalStore { |
|
161 |
+ |
/// Open the store at `root`, creating the directory if needed. |
|
162 |
+ |
pub fn open(root: impl Into<PathBuf>) -> Result<LocalStore> { |
|
163 |
+ |
let root = root.into(); |
|
164 |
+ |
fs::create_dir_all(&root)?; |
|
165 |
+ |
Ok(LocalStore { root }) |
|
166 |
+ |
} |
|
167 |
+ |
|
|
168 |
+ |
pub fn root(&self) -> &Path { |
|
169 |
+ |
&self.root |
|
170 |
+ |
} |
|
171 |
+ |
|
|
172 |
+ |
fn path(&self, key: &str) -> Result<PathBuf> { |
|
173 |
+ |
check_key(key)?; |
|
174 |
+ |
Ok(self.root.join(key)) |
|
175 |
+ |
} |
|
176 |
+ |
|
|
177 |
+ |
fn temp_beside(&self, path: &Path) -> PathBuf { |
|
178 |
+ |
let mut nonce = [0u8; 6]; |
|
179 |
+ |
rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut nonce); |
|
180 |
+ |
let name = path |
|
181 |
+ |
.file_name() |
|
182 |
+ |
.map(|n| n.to_string_lossy().into_owned()) |
|
183 |
+ |
.unwrap_or_default(); |
|
184 |
+ |
path.with_file_name(format!(".{name}.tmp-{}", hex::encode(nonce))) |
|
185 |
+ |
} |
|
186 |
+ |
|
|
187 |
+ |
fn list_dir(&self, dir: &Path, rel: &str, prefix: &str, out: &mut Vec<Entry>) -> Result<()> { |
|
188 |
+ |
let entries = match fs::read_dir(dir) { |
|
189 |
+ |
Ok(e) => e, |
|
190 |
+ |
Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()), |
|
191 |
+ |
Err(e) => return Err(e.into()), |
|
192 |
+ |
}; |
|
193 |
+ |
for entry in entries { |
|
194 |
+ |
let entry = entry?; |
|
195 |
+ |
let name = entry.file_name().to_string_lossy().into_owned(); |
|
196 |
+ |
// Temporary and lock files are not objects. |
|
197 |
+ |
if name.starts_with('.') { |
|
198 |
+ |
continue; |
|
199 |
+ |
} |
|
200 |
+ |
let key = if rel.is_empty() { |
|
201 |
+ |
name.clone() |
|
202 |
+ |
} else { |
|
203 |
+ |
format!("{rel}/{name}") |
|
204 |
+ |
}; |
|
205 |
+ |
let ty = entry.file_type()?; |
|
206 |
+ |
if ty.is_dir() { |
|
207 |
+ |
// Descend only where the prefix could still match. |
|
208 |
+ |
let dir_prefix = format!("{key}/"); |
|
209 |
+ |
if prefix.starts_with(&dir_prefix) || dir_prefix.starts_with(prefix) { |
|
210 |
+ |
self.list_dir(&entry.path(), &key, prefix, out)?; |
|
211 |
+ |
} |
|
212 |
+ |
} else if ty.is_file() && key.starts_with(prefix) { |
|
213 |
+ |
out.push(Entry { |
|
214 |
+ |
key, |
|
215 |
+ |
size: entry.metadata()?.len(), |
|
216 |
+ |
}); |
|
217 |
+ |
} |
|
218 |
+ |
} |
|
219 |
+ |
Ok(()) |
|
220 |
+ |
} |
|
221 |
+ |
|
|
222 |
+ |
/// Take `path.lock`, waiting for another writer, and removing a lock |
|
223 |
+ |
/// a dead process left behind. |
|
224 |
+ |
fn lock(&self, key: &str, path: &Path) -> Result<PathBuf> { |
|
225 |
+ |
let lock = path.with_file_name(format!( |
|
226 |
+ |
".{}.lock", |
|
227 |
+ |
path.file_name() |
|
228 |
+ |
.map(|n| n.to_string_lossy().into_owned()) |
|
229 |
+ |
.unwrap_or_default() |
|
230 |
+ |
)); |
|
231 |
+ |
let started = Instant::now(); |
|
232 |
+ |
loop { |
|
233 |
+ |
match fs::OpenOptions::new() |
|
234 |
+ |
.write(true) |
|
235 |
+ |
.create_new(true) |
|
236 |
+ |
.open(&lock) |
|
237 |
+ |
{ |
|
238 |
+ |
Ok(mut f) => { |
|
239 |
+ |
let _ = writeln!(f, "{}", std::process::id()); |
|
240 |
+ |
return Ok(lock); |
|
241 |
+ |
} |
|
242 |
+ |
Err(e) if e.kind() == io::ErrorKind::AlreadyExists => { |
|
243 |
+ |
if let Ok(m) = fs::metadata(&lock) { |
|
244 |
+ |
let age = m.modified().ok().and_then(|t| t.elapsed().ok()); |
|
245 |
+ |
if age.is_some_and(|a| a > LOCK_STALE) { |
|
246 |
+ |
let _ = fs::remove_file(&lock); |
|
247 |
+ |
continue; |
|
248 |
+ |
} |
|
249 |
+ |
} |
|
250 |
+ |
if started.elapsed() > LOCK_WAIT { |
|
251 |
+ |
return Err(Error::Busy(key.to_string())); |
|
252 |
+ |
} |
|
253 |
+ |
std::thread::sleep(Duration::from_millis(10)); |
|
254 |
+ |
} |
|
255 |
+ |
Err(e) => return Err(e.into()), |
|
256 |
+ |
} |
|
257 |
+ |
} |
|
258 |
+ |
} |
|
259 |
+ |
} |
|
260 |
+ |
|
|
261 |
+ |
impl Store for LocalStore { |
|
262 |
+ |
fn describe(&self) -> String { |
|
263 |
+ |
self.root.display().to_string() |
|
264 |
+ |
} |
|
265 |
+ |
|
|
266 |
+ |
fn has(&self, key: &str) -> Result<bool> { |
|
267 |
+ |
Ok(self.path(key)?.is_file()) |
|
268 |
+ |
} |
|
269 |
+ |
|
|
270 |
+ |
fn get(&self, key: &str) -> Result<Vec<u8>> { |
|
271 |
+ |
let path = self.path(key)?; |
|
272 |
+ |
match fs::read(&path) { |
|
273 |
+ |
Ok(b) => Ok(b), |
|
274 |
+ |
Err(e) if e.kind() == io::ErrorKind::NotFound => Err(Error::NotFound(key.to_string())), |
|
275 |
+ |
Err(e) => Err(e.into()), |
|
276 |
+ |
} |
|
277 |
+ |
} |
|
278 |
+ |
|
|
279 |
+ |
fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> { |
|
280 |
+ |
let path = self.path(key)?; |
|
281 |
+ |
let mut f = match fs::File::open(&path) { |
|
282 |
+ |
Ok(f) => f, |
|
283 |
+ |
Err(e) if e.kind() == io::ErrorKind::NotFound => { |
|
284 |
+ |
return Err(Error::NotFound(key.to_string())) |
|
285 |
+ |
} |
|
286 |
+ |
Err(e) => return Err(e.into()), |
|
287 |
+ |
}; |
|
288 |
+ |
let len = f.metadata()?.len(); |
|
289 |
+ |
if range.end > len || range.start > range.end { |
|
290 |
+ |
return Err(Error::Backend(format!( |
|
291 |
+ |
"{key}: bytes {}..{} asked of an object of {len} bytes", |
|
292 |
+ |
range.start, range.end |
|
293 |
+ |
))); |
|
294 |
+ |
} |
|
295 |
+ |
f.seek(SeekFrom::Start(range.start))?; |
|
296 |
+ |
let mut out = vec![0u8; (range.end - range.start) as usize]; |
|
297 |
+ |
f.read_exact(&mut out)?; |
|
298 |
+ |
Ok(out) |
|
299 |
+ |
} |
|
300 |
+ |
|
|
301 |
+ |
fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> { |
|
302 |
+ |
let path = self.path(key)?; |
|
303 |
+ |
if path.exists() { |
|
304 |
+ |
return Err(Error::Exists(key.to_string())); |
|
305 |
+ |
} |
|
306 |
+ |
if let Some(dir) = path.parent() { |
|
307 |
+ |
fs::create_dir_all(dir)?; |
|
308 |
+ |
} |
|
309 |
+ |
// Written beside its name, then linked into it: the link fails if |
|
310 |
+ |
// the name was taken meanwhile, and a reader never sees a part. |
|
311 |
+ |
let tmp = self.temp_beside(&path); |
|
312 |
+ |
let written = (|| { |
|
313 |
+ |
let mut f = fs::File::create(&tmp)?; |
|
314 |
+ |
f.write_all(bytes)?; |
|
315 |
+ |
f.sync_all() |
|
316 |
+ |
})(); |
|
317 |
+ |
let linked = written.and_then(|()| fs::hard_link(&tmp, &path)); |
|
318 |
+ |
let _ = fs::remove_file(&tmp); |
|
319 |
+ |
match linked { |
|
320 |
+ |
Ok(()) => { |
|
321 |
+ |
if let Some(dir) = path.parent() { |
|
322 |
+ |
crate::durable::sync_dir(dir)?; |
|
323 |
+ |
} |
|
324 |
+ |
Ok(()) |
|
325 |
+ |
} |
|
326 |
+ |
Err(e) if e.kind() == io::ErrorKind::AlreadyExists => { |
|
327 |
+ |
Err(Error::Exists(key.to_string())) |
|
328 |
+ |
} |
|
329 |
+ |
Err(e) => Err(e.into()), |
|
330 |
+ |
} |
|
331 |
+ |
} |
|
332 |
+ |
|
|
333 |
+ |
fn list(&self, prefix: &str) -> Result<Vec<Entry>> { |
|
334 |
+ |
let mut out = Vec::new(); |
|
335 |
+ |
self.list_dir(&self.root, "", prefix, &mut out)?; |
|
336 |
+ |
out.sort_by(|a, b| a.key.cmp(&b.key)); |
|
337 |
+ |
Ok(out) |
|
338 |
+ |
} |
|
339 |
+ |
|
|
340 |
+ |
fn delete(&self, key: &str) -> Result<()> { |
|
341 |
+ |
let path = self.path(key)?; |
|
342 |
+ |
match fs::remove_file(&path) { |
|
343 |
+ |
Ok(()) => Ok(()), |
|
344 |
+ |
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()), |
|
345 |
+ |
Err(e) => Err(e.into()), |
|
346 |
+ |
} |
|
347 |
+ |
} |
|
348 |
+ |
|
|
349 |
+ |
fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> { |
|
350 |
+ |
let path = self.path(key)?; |
|
351 |
+ |
match fs::read(&path) { |
|
352 |
+ |
Ok(b) => { |
|
353 |
+ |
let v = content_version(&b); |
|
354 |
+ |
Ok(Some((b, v))) |
|
355 |
+ |
} |
|
356 |
+ |
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None), |
|
357 |
+ |
Err(e) => Err(e.into()), |
|
358 |
+ |
} |
|
359 |
+ |
} |
|
360 |
+ |
|
|
361 |
+ |
fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> { |
|
362 |
+ |
let path = self.path(key)?; |
|
363 |
+ |
if let Some(dir) = path.parent() { |
|
364 |
+ |
fs::create_dir_all(dir)?; |
|
365 |
+ |
} |
|
366 |
+ |
let lock = self.lock(key, &path)?; |
|
367 |
+ |
let result = (|| { |
|
368 |
+ |
let current = self.read_ref(key)?.map(|(_, v)| v); |
|
369 |
+ |
check_expected(key, current.as_ref(), expected)?; |
|
370 |
+ |
crate::durable::write_atomic(&path, bytes)?; |
|
371 |
+ |
Ok(content_version(bytes)) |
|
372 |
+ |
})(); |
|
373 |
+ |
let _ = fs::remove_file(&lock); |
|
374 |
+ |
result |
|
375 |
+ |
} |
|
376 |
+ |
} |
|
377 |
+ |
|
|
378 |
+ |
// ---------------------------------------------------------------- memory |
|
379 |
+ |
|
|
380 |
+ |
/// A store in memory, for tests: the conditional writes behave as the |
|
381 |
+ |
/// cloud stores' do. |
|
382 |
+ |
#[derive(Default)] |
|
383 |
+ |
pub struct MemoryStore { |
|
384 |
+ |
objects: Mutex<BTreeMap<String, Vec<u8>>>, |
|
385 |
+ |
} |
|
386 |
+ |
|
|
387 |
+ |
impl MemoryStore { |
|
388 |
+ |
pub fn new() -> MemoryStore { |
|
389 |
+ |
MemoryStore::default() |
|
390 |
+ |
} |
|
391 |
+ |
|
|
392 |
+ |
fn map(&self) -> std::sync::MutexGuard<'_, BTreeMap<String, Vec<u8>>> { |
|
393 |
+ |
self.objects.lock().unwrap_or_else(|p| p.into_inner()) |
|
394 |
+ |
} |
|
395 |
+ |
} |
|
396 |
+ |
|
|
397 |
+ |
impl Store for MemoryStore { |
|
398 |
+ |
fn describe(&self) -> String { |
|
399 |
+ |
"memory".into() |
|
400 |
+ |
} |
|
401 |
+ |
|
|
402 |
+ |
fn has(&self, key: &str) -> Result<bool> { |
|
403 |
+ |
check_key(key)?; |
|
404 |
+ |
Ok(self.map().contains_key(key)) |
|
405 |
+ |
} |
|
406 |
+ |
|
|
407 |
+ |
fn get(&self, key: &str) -> Result<Vec<u8>> { |
|
408 |
+ |
check_key(key)?; |
|
409 |
+ |
self.map() |
|
410 |
+ |
.get(key) |
|
411 |
+ |
.cloned() |
|
412 |
+ |
.ok_or_else(|| Error::NotFound(key.to_string())) |
|
413 |
+ |
} |
|
414 |
+ |
|
|
415 |
+ |
fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> { |
|
416 |
+ |
let all = self.get(key)?; |
|
417 |
+ |
let len = all.len() as u64; |
|
418 |
+ |
if range.end > len || range.start > range.end { |
|
419 |
+ |
return Err(Error::Backend(format!( |
|
420 |
+ |
"{key}: bytes {}..{} asked of an object of {len} bytes", |
|
421 |
+ |
range.start, range.end |
|
422 |
+ |
))); |
|
423 |
+ |
} |
|
424 |
+ |
Ok(all[range.start as usize..range.end as usize].to_vec()) |
|
425 |
+ |
} |
|
426 |
+ |
|
|
427 |
+ |
fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> { |
|
428 |
+ |
check_key(key)?; |
|
429 |
+ |
let mut map = self.map(); |
|
430 |
+ |
if map.contains_key(key) { |
|
431 |
+ |
return Err(Error::Exists(key.to_string())); |
|
432 |
+ |
} |
|
433 |
+ |
map.insert(key.to_string(), bytes.to_vec()); |
|
434 |
+ |
Ok(()) |
|
435 |
+ |
} |
|
436 |
+ |
|
|
437 |
+ |
fn list(&self, prefix: &str) -> Result<Vec<Entry>> { |
|
438 |
+ |
Ok(self |
|
439 |
+ |
.map() |
|
440 |
+ |
.range(prefix.to_string()..) |
|
441 |
+ |
.take_while(|(k, _)| k.starts_with(prefix)) |
|
442 |
+ |
.map(|(k, v)| Entry { |
|
443 |
+ |
key: k.clone(), |
|
444 |
+ |
size: v.len() as u64, |
|
445 |
+ |
}) |
|
446 |
+ |
.collect()) |
|
447 |
+ |
} |
|
448 |
+ |
|
|
449 |
+ |
fn delete(&self, key: &str) -> Result<()> { |
|
450 |
+ |
check_key(key)?; |
|
451 |
+ |
self.map().remove(key); |
|
452 |
+ |
Ok(()) |
|
453 |
+ |
} |
|
454 |
+ |
|
|
455 |
+ |
fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> { |
|
456 |
+ |
check_key(key)?; |
|
457 |
+ |
Ok(self.map().get(key).map(|b| (b.clone(), content_version(b)))) |
|
458 |
+ |
} |
|
459 |
+ |
|
|
460 |
+ |
fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> { |
|
461 |
+ |
check_key(key)?; |
|
462 |
+ |
let mut map = self.map(); |
|
463 |
+ |
let current = map.get(key).map(|b| content_version(b)); |
|
464 |
+ |
check_expected(key, current.as_ref(), expected)?; |
|
465 |
+ |
map.insert(key.to_string(), bytes.to_vec()); |
|
466 |
+ |
Ok(content_version(bytes)) |
|
467 |
+ |
} |
|
468 |
+ |
} |
|
469 |
+ |
|
|
470 |
+ |
// ---------------------------------------------------------------- conformance |
|
471 |
+ |
|
|
472 |
+ |
/// Exercise a store under `prefix` (which must be free, and is left |
|
473 |
+ |
/// empty) and report what passed: object round trips, ranges, create-only |
|
474 |
+ |
/// refusal, listing, deletion, and compare-and-swap on a reference. Every |
|
475 |
+ |
/// store runs this in its tests, and `oxy remote check` runs it against a |
|
476 |
+ |
/// remote before trusting it with a repository. |
|
477 |
+ |
pub fn conformance(store: &dyn Store, prefix: &str) -> Result<Vec<&'static str>> { |
|
478 |
+ |
let mut passed = Vec::new(); |
|
479 |
+ |
let key = |name: &str| format!("{prefix}/{name}"); |
|
480 |
+ |
let fail = |what: &str| Error::Backend(format!("{}: {what}", store.describe())); |
|
481 |
+ |
let expect = |ok: bool, what: &str| if ok { Ok(()) } else { Err(fail(what)) }; |
|
482 |
+ |
|
|
483 |
+ |
expect( |
|
484 |
+ |
store.list(prefix)?.is_empty(), |
|
485 |
+ |
"the probe prefix is not empty", |
|
486 |
+ |
)?; |
|
487 |
+ |
expect( |
|
488 |
+ |
!store.has(&key("a"))?, |
|
489 |
+ |
"has() of a missing object says true", |
|
490 |
+ |
)?; |
|
491 |
+ |
expect( |
|
492 |
+ |
matches!(store.get(&key("a")), Err(Error::NotFound(_))), |
|
493 |
+ |
"get() of a missing object is not NotFound", |
|
494 |
+ |
)?; |
|
495 |
+ |
passed.push("missing objects are reported missing"); |
|
496 |
+ |
|
|
497 |
+ |
let a: Vec<u8> = (0..256u32 * 1024).map(|i| (i % 251) as u8).collect(); |
|
498 |
+ |
store.put_new(&key("a"), &a)?; |
|
499 |
+ |
expect(store.has(&key("a"))?, "has() after put_new says false")?; |
|
500 |
+ |
expect( |
|
501 |
+ |
store.get(&key("a"))? == a, |
|
502 |
+ |
"get() returns other bytes than put_new wrote", |
|
503 |
+ |
)?; |
|
504 |
+ |
passed.push("an object round-trips"); |
|
505 |
+ |
|
|
506 |
+ |
expect( |
|
507 |
+ |
store.get_range(&key("a"), 1000..1500)? == a[1000..1500], |
|
508 |
+ |
"get_range() returns the wrong bytes", |
|
509 |
+ |
)?; |
|
510 |
+ |
expect( |
|
511 |
+ |
store.get_range(&key("a"), 0..0)?.is_empty(), |
|
512 |
+ |
"an empty range is not empty", |
|
513 |
+ |
)?; |
|
514 |
+ |
expect( |
|
515 |
+ |
store.get_range(&key("a"), 0..a.len() as u64 + 1).is_err(), |
|
516 |
+ |
"a range past the end is not refused", |
|
517 |
+ |
)?; |
|
518 |
+ |
passed.push("ranges read correctly and are bounded"); |
|
519 |
+ |
|
|
520 |
+ |
expect( |
|
521 |
+ |
matches!(store.put_new(&key("a"), b"other"), Err(Error::Exists(_))), |
|
522 |
+ |
"put_new() over an existing object is not refused", |
|
523 |
+ |
)?; |
|
524 |
+ |
expect( |
|
525 |
+ |
store.get(&key("a"))? == a, |
|
526 |
+ |
"a refused put_new() changed the object", |
|
527 |
+ |
)?; |
|
528 |
+ |
passed.push("objects are written once"); |
|
529 |
+ |
|
|
530 |
+ |
store.put_new(&key("dir/b"), b"bb")?; |
|
531 |
+ |
let listed = store.list(prefix)?; |
|
532 |
+ |
expect( |
|
533 |
+ |
listed |
|
534 |
+ |
== vec![ |
|
535 |
+ |
Entry { |
|
536 |
+ |
key: key("a"), |
|
537 |
+ |
size: a.len() as u64, |
|
538 |
+ |
}, |
|
539 |
+ |
Entry { |
|
540 |
+ |
key: key("dir/b"), |
|
541 |
+ |
size: 2, |
|
542 |
+ |
}, |
|
543 |
+ |
], |
|
544 |
+ |
"list() does not return the two objects, sorted, with their sizes", |
|
545 |
+ |
)?; |
|
546 |
+ |
expect( |
|
547 |
+ |
store.list(&key("dir/"))? |
|
548 |
+ |
== vec![Entry { |
|
549 |
+ |
key: key("dir/b"), |
|
550 |
+ |
size: 2, |
|
551 |
+ |
}], |
|
552 |
+ |
"list() of a sub-prefix is wrong", |
|
553 |
+ |
)?; |
|
554 |
+ |
expect( |
|
555 |
+ |
store.list(&key("nothing/"))?.is_empty(), |
|
556 |
+ |
"list() of an unused prefix is not empty", |
|
557 |
+ |
)?; |
|
558 |
+ |
passed.push("listing by prefix"); |
|
559 |
+ |
|
|
560 |
+ |
store.delete(&key("dir/b"))?; |
|
561 |
+ |
expect( |
|
562 |
+ |
!store.has(&key("dir/b"))?, |
|
563 |
+ |
"an object is still there after delete()", |
|
564 |
+ |
)?; |
|
565 |
+ |
store.delete(&key("dir/b"))?; |
|
566 |
+ |
passed.push("deletion, and deleting again, succeed"); |
|
567 |
+ |
|
|
568 |
+ |
let r = key("refs"); |
|
569 |
+ |
expect( |
|
570 |
+ |
store.read_ref(&r)?.is_none(), |
|
571 |
+ |
"an unwritten reference reads as something", |
|
572 |
+ |
)?; |
|
573 |
+ |
let v1 = store.write_ref(&r, b"one", None)?; |
|
574 |
+ |
expect( |
|
575 |
+ |
matches!(store.write_ref(&r, b"again", None), Err(Error::Conflict(_))), |
|
576 |
+ |
"a second create-only write of the reference is not refused", |
|
577 |
+ |
)?; |
|
578 |
+ |
let (bytes, seen) = store |
|
579 |
+ |
.read_ref(&r)? |
|
580 |
+ |
.ok_or_else(|| fail("the reference vanished"))?; |
|
581 |
+ |
expect( |
|
582 |
+ |
bytes == b"one" && seen == v1, |
|
583 |
+ |
"the reference reads back wrong", |
|
584 |
+ |
)?; |
|
585 |
+ |
let v2 = store.write_ref(&r, b"two", Some(&v1))?; |
|
586 |
+ |
expect(v2 != v1, "the version did not change on write")?; |
|
587 |
+ |
expect( |
|
588 |
+ |
matches!( |
|
589 |
+ |
store.write_ref(&r, b"three", Some(&v1)), |
|
590 |
+ |
Err(Error::Conflict(_)) |
|
591 |
+ |
), |
|
592 |
+ |
"a write with a stale version is not refused", |
|
593 |
+ |
)?; |
|
594 |
+ |
let (bytes, seen) = store |
|
595 |
+ |
.read_ref(&r)? |
|
596 |
+ |
.ok_or_else(|| fail("the reference vanished"))?; |
|
597 |
+ |
expect( |
|
598 |
+ |
bytes == b"two" && seen == v2, |
|
599 |
+ |
"a refused write changed the reference", |
|
600 |
+ |
)?; |
|
601 |
+ |
passed.push("references update by compare-and-swap"); |
|
602 |
+ |
|
|
603 |
+ |
store.delete(&key("a"))?; |
|
604 |
+ |
store.delete(&r)?; |
|
605 |
+ |
expect( |
|
606 |
+ |
store.list(prefix)?.is_empty(), |
|
607 |
+ |
"the probe prefix is not empty after cleanup", |
|
608 |
+ |
)?; |
|
609 |
+ |
passed.push("cleanup"); |
|
610 |
+ |
Ok(passed) |
|
611 |
+ |
} |
|
612 |
+ |
|
|
613 |
+ |
#[cfg(test)] |
|
614 |
+ |
mod tests { |
|
615 |
+ |
use super::*; |
|
616 |
+ |
use std::sync::Arc; |
|
617 |
+ |
|
|
618 |
+ |
#[test] |
|
619 |
+ |
fn keys_are_plain_relative_paths() { |
|
620 |
+ |
for ok in ["a", "packs/ab.pack", "a-b_c.d/e"] { |
|
621 |
+ |
check_key(ok).unwrap(); |
|
622 |
+ |
} |
|
623 |
+ |
for bad in [ |
|
624 |
+ |
"", "/a", "a/", "a//b", "..", "a/../b", "./a", "a b", "a\\b", "ä", |
|
625 |
+ |
] { |
|
626 |
+ |
assert!(check_key(bad).is_err(), "{bad:?}"); |
|
627 |
+ |
} |
|
628 |
+ |
assert!(check_key(&"x".repeat(MAX_KEY_LEN + 1)).is_err()); |
|
629 |
+ |
} |
|
630 |
+ |
|
|
631 |
+ |
#[test] |
|
632 |
+ |
fn the_memory_store_conforms() { |
|
633 |
+ |
let store = MemoryStore::new(); |
|
634 |
+ |
let passed = conformance(&store, "probe").unwrap(); |
|
635 |
+ |
assert_eq!(passed.len(), 8, "{passed:?}"); |
|
636 |
+ |
} |
|
637 |
+ |
|
|
638 |
+ |
#[test] |
|
639 |
+ |
fn the_directory_store_conforms() { |
|
640 |
+ |
let dir = tempfile::tempdir().unwrap(); |
|
641 |
+ |
let store = LocalStore::open(dir.path().join("bucket")).unwrap(); |
|
642 |
+ |
let passed = conformance(&store, "repos/one").unwrap(); |
|
643 |
+ |
assert_eq!(passed.len(), 8, "{passed:?}"); |
|
644 |
+ |
// Nothing but the (empty) directories and no stray temporaries. |
|
645 |
+ |
let stray: Vec<_> = walkdir(dir.path()) |
|
646 |
+ |
.into_iter() |
|
647 |
+ |
.filter(|p| p.is_file()) |
|
648 |
+ |
.collect(); |
|
649 |
+ |
assert!(stray.is_empty(), "{stray:?}"); |
|
650 |
+ |
} |
|
651 |
+ |
|
|
652 |
+ |
fn walkdir(dir: &Path) -> Vec<PathBuf> { |
|
653 |
+ |
let mut out = vec![]; |
|
654 |
+ |
if let Ok(rd) = fs::read_dir(dir) { |
|
655 |
+ |
for e in rd.flatten() { |
|
656 |
+ |
let p = e.path(); |
|
657 |
+ |
if p.is_dir() { |
|
658 |
+ |
out.extend(walkdir(&p)); |
|
659 |
+ |
} else { |
|
660 |
+ |
out.push(p); |
|
661 |
+ |
} |
|
662 |
+ |
} |
|
663 |
+ |
} |
|
664 |
+ |
out |
|
665 |
+ |
} |
|
666 |
+ |
|
|
667 |
+ |
#[test] |
|
668 |
+ |
fn a_directory_store_refuses_keys_that_leave_its_root() { |
|
669 |
+ |
let dir = tempfile::tempdir().unwrap(); |
|
670 |
+ |
let store = LocalStore::open(dir.path().join("bucket")).unwrap(); |
|
671 |
+ |
assert!(matches!( |
|
672 |
+ |
store.put_new("../escape", b"x"), |
|
673 |
+ |
Err(Error::Key(..)) |
|
674 |
+ |
)); |
|
675 |
+ |
assert!(matches!(store.get("/etc/passwd"), Err(Error::Key(..)))); |
|
676 |
+ |
assert!(!dir.path().join("escape").exists()); |
|
677 |
+ |
} |
|
678 |
+ |
|
|
679 |
+ |
/// Many threads race to move a reference from the same version: exactly |
|
680 |
+ |
/// one wins each round, on both stores. |
|
681 |
+ |
#[test] |
|
682 |
+ |
fn compare_and_swap_admits_one_writer_per_round() { |
|
683 |
+ |
let dir = tempfile::tempdir().unwrap(); |
|
684 |
+ |
let local: Arc<dyn Store> = Arc::new(LocalStore::open(dir.path()).unwrap()); |
|
685 |
+ |
let memory: Arc<dyn Store> = Arc::new(MemoryStore::new()); |
|
686 |
+ |
for store in [local, memory] { |
|
687 |
+ |
let v0 = store.write_ref("refs", b"0", None).unwrap(); |
|
688 |
+ |
let wins: Vec<bool> = (0..8) |
|
689 |
+ |
.map(|i| { |
|
690 |
+ |
let store = store.clone(); |
|
691 |
+ |
let v0 = v0.clone(); |
|
692 |
+ |
std::thread::spawn(move || { |
|
693 |
+ |
store |
|
694 |
+ |
.write_ref("refs", format!("{}", i + 1).as_bytes(), Some(&v0)) |
|
695 |
+ |
.is_ok() |
|
696 |
+ |
}) |
|
697 |
+ |
}) |
|
698 |
+ |
.collect::<Vec<_>>() |
|
699 |
+ |
.into_iter() |
|
700 |
+ |
.map(|h| h.join().unwrap()) |
|
701 |
+ |
.collect(); |
|
702 |
+ |
assert_eq!( |
|
703 |
+ |
wins.iter().filter(|w| **w).count(), |
|
704 |
+ |
1, |
|
705 |
+ |
"{}: {wins:?}", |
|
706 |
+ |
store.describe() |
|
707 |
+ |
); |
|
708 |
+ |
let (bytes, _) = store.read_ref("refs").unwrap().unwrap(); |
|
709 |
+ |
assert_ne!(bytes, b"0"); |
|
710 |
+ |
} |
|
711 |
+ |
} |
|
712 |
+ |
|
|
713 |
+ |
#[test] |
|
714 |
+ |
fn a_stale_lock_is_broken_and_a_live_one_waited_for() { |
|
715 |
+ |
let dir = tempfile::tempdir().unwrap(); |
|
716 |
+ |
let store = LocalStore::open(dir.path()).unwrap(); |
|
717 |
+ |
let lock = dir.path().join(".refs.lock"); |
|
718 |
+ |
fs::write(&lock, "dead").unwrap(); |
|
719 |
+ |
let old = std::time::SystemTime::now() - LOCK_STALE - Duration::from_secs(5); |
|
720 |
+ |
fs::File::open(&lock).unwrap().set_modified(old).unwrap(); |
|
721 |
+ |
store.write_ref("refs", b"after a crash", None).unwrap(); |
|
722 |
+ |
assert!(!lock.exists()); |
|
723 |
+ |
} |
|
724 |
+ |
|
|
725 |
+ |
#[test] |
|
726 |
+ |
fn temporaries_and_locks_are_not_listed() { |
|
727 |
+ |
let dir = tempfile::tempdir().unwrap(); |
|
728 |
+ |
let store = LocalStore::open(dir.path()).unwrap(); |
|
729 |
+ |
store.put_new("packs/a", b"a").unwrap(); |
|
730 |
+ |
fs::write(dir.path().join("packs/.a.tmp-ffffff"), "junk").unwrap(); |
|
731 |
+ |
fs::write(dir.path().join(".refs.lock"), "1").unwrap(); |
|
732 |
+ |
let keys: Vec<String> = store.list("").unwrap().into_iter().map(|e| e.key).collect(); |
|
733 |
+ |
assert_eq!(keys, ["packs/a"]); |
|
734 |
+ |
} |
|
735 |
+ |
} |