This repository has no description
20 kB
612 lines
1//! The lower layer: all store paths of OUT/paths.json, read-only. Metadata
2//! comes from the caches' `.ls` listings, all fetched up front and kept in
3//! memory; the first open() of a file in a store path fetches, verifies
4//! (NarHash/NarSize) and unpacks that path's NAR. Opens of a path that is
5//! being fetched wait for that fetch.
6
7use std::collections::{BTreeMap, HashMap};
8use std::ffi::{CStr, CString};
9use std::fs::File;
10use std::io;
11use std::mem::ManuallyDrop;
12use std::os::fd::{AsRawFd, FromRawFd};
13use std::path::{Path, PathBuf};
14use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
15use std::sync::{Arc, Mutex, RwLock};
16use std::time::{Duration, Instant};
17
18use fuse_backend_rs::abi::fuse_abi::stat64;
19use fuse_backend_rs::api::filesystem::{
20 Context, DirEntry, Entry, FileSystem, GetxattrReply, ListxattrReply, OpenOptions,
21 ZeroCopyWriter,
22};
23use harmonia_utils_base_encoding::base32;
24
25use crate::closure::{par_map, PathInfo};
26use crate::http::{self, Http};
27use crate::nar::{self, CountingReader, Tree};
28use crate::stats::Stats;
29use crate::Backing;
30
31/// The store paths never change: let the kernel cache everything.
32pub const TTL: Duration = Duration::from_secs(86400);
33
34/// Set on the inode numbers (st_ino) of this layer, so they don't collide
35/// with the upper layer's.
36pub const INO_TAG: u64 = 1 << 62;
37
38enum Kind {
39 /// children sorted by name
40 Dir(Vec<(Box<str>, u32)>),
41 File {
42 size: u64,
43 exec: bool,
44 },
45 Symlink(Box<[u8]>),
46}
47
48struct Node {
49 kind: Kind,
50 /// index into `LazyFs::names`
51 path: u32,
52 /// path below the store path (empty for its root)
53 rel: Box<str>,
54}
55
56struct Handle {
57 file: File,
58 ino: u64,
59}
60
61pub struct LazyFs {
62 names: Vec<String>,
63 infos: Vec<PathInfo>,
64 /// inode `i` is `nodes[i - 1]`; inode 1 is the store dir
65 nodes: Vec<Node>,
66 cache: PathBuf,
67 http: Http,
68 pub stats: Arc<Stats>,
69 /// per store path: where it is unpacked, once fetched
70 fetched: Vec<Mutex<Option<PathBuf>>>,
71 handles: RwLock<HashMap<u64, Arc<Handle>>>,
72 next_handle: AtomicU64,
73 backing: Option<Arc<Backing>>,
74 passthrough: AtomicBool,
75 /// backing id and open count per inode: the kernel wants all
76 /// passthrough opens of an inode to use the same backing file
77 backing_ids: Mutex<HashMap<u64, (u32, usize)>>,
78 backing_failed: AtomicBool,
79}
80
81fn err(e: i32) -> io::Error {
82 io::Error::from_raw_os_error(e)
83}
84
85impl LazyFs {
86 pub fn new(outdir: &Path, backing: Option<Arc<Backing>>) -> io::Result<LazyFs> {
87 let paths: BTreeMap<String, PathInfo> =
88 serde_json::from_slice(&std::fs::read(outdir.join("paths.json"))?)?;
89 let (names, infos): (Vec<String>, Vec<PathInfo>) = paths.into_iter().unzip();
90 let cache = outdir.join("cache");
91 std::fs::create_dir_all(&cache)?;
92 let http = Http::new(infos.iter().map(|i| i.cache.as_str()));
93 let mut fs = LazyFs {
94 fetched: names.iter().map(|_| Mutex::new(None)).collect(),
95 names,
96 infos,
97 nodes: Vec::new(),
98 cache,
99 http,
100 stats: Arc::new(Stats::new(Some(outdir.join("stats.json")))),
101 handles: RwLock::new(HashMap::new()),
102 next_handle: AtomicU64::new(1),
103 backing,
104 passthrough: AtomicBool::new(false),
105 backing_ids: Mutex::new(HashMap::new()),
106 backing_failed: AtomicBool::new(false),
107 };
108 fs.load_listings()?;
109 fs.stats.save();
110 Ok(fs)
111 }
112
113 /// Fetch the `.ls` listing of every store path (falling back to the NAR
114 /// for paths without one) and build the inode table from them. Fails if
115 /// any path is left without a listing: all of them get registered as
116 /// valid in nix's database.
117 fn load_listings(&mut self) -> io::Result<()> {
118 let t = Instant::now();
119 let idx: Vec<usize> = (0..self.names.len()).collect();
120 let done = AtomicUsize::new(0);
121 let trees = par_map(&idx, 64, |&i| {
122 let r = self.listing(i);
123 let n = done.fetch_add(1, Ordering::Relaxed) + 1;
124 if n.is_multiple_of(500) {
125 eprintln!("listings: {n}/{}", idx.len());
126 }
127 r
128 });
129 let mut root = Vec::with_capacity(trees.len());
130 self.nodes.push(Node {
131 kind: Kind::Dir(Vec::new()),
132 path: u32::MAX,
133 rel: "".into(),
134 });
135 let mut failed = 0;
136 for (i, t) in trees.into_iter().enumerate() {
137 match t {
138 Ok(t) => {
139 let ino = self.add(i as u32, String::new(), t);
140 root.push((self.names[i].clone().into_boxed_str(), ino));
141 }
142 Err(e) => {
143 eprintln!("no listing for {}: {e}", self.names[i]);
144 failed += 1;
145 }
146 }
147 }
148 if failed > 0 {
149 return Err(io::Error::other(format!(
150 "{failed} of {} store paths without a listing",
151 self.names.len()
152 )));
153 }
154 root.sort();
155 self.nodes[0].kind = Kind::Dir(root);
156 eprintln!(
157 "listings: {} store paths, {} inodes in {:.1} s",
158 self.names.len(),
159 self.nodes.len(),
160 t.elapsed().as_secs_f64(),
161 );
162 Ok(())
163 }
164
165 fn listing(&self, i: usize) -> io::Result<Tree> {
166 let info = &self.infos[i];
167 let h = self.names[i].split('-').next().unwrap_or("");
168 match self.http.get(&format!("{}/{h}.ls", info.cache)) {
169 Ok((body, len)) => {
170 self.stats.add("ls", len as u64);
171 Tree::from_listing(&body)
172 }
173 // no listing: fall back to the NAR
174 Err(http::Error::NotFound) => Tree::from_path(&self.fetch(i)?),
175 Err(e) => Err(e.into()),
176 }
177 }
178
179 /// Add the nodes of `t` (at `rel` in store path `path`), return its inode.
180 fn add(&mut self, path: u32, rel: String, t: Tree) -> u32 {
181 let ino = self.nodes.len() as u32 + 1;
182 let (kind, entries) = match t {
183 Tree::File { size, exec } => (Kind::File { size, exec }, None),
184 Tree::Symlink(s) => (Kind::Symlink(s.into_bytes().into()), None),
185 Tree::Dir(e) => (Kind::Dir(Vec::new()), Some(e)),
186 };
187 self.nodes.push(Node {
188 kind,
189 path,
190 rel: rel.clone().into(),
191 });
192 if let Some(entries) = entries {
193 let mut children = Vec::with_capacity(entries.len());
194 for (name, c) in entries {
195 let r = if rel.is_empty() {
196 name.clone()
197 } else {
198 format!("{rel}/{name}")
199 };
200 children.push((name.into_boxed_str(), self.add(path, r, c)));
201 }
202 // BTreeMap order is sorted already
203 self.nodes[ino as usize - 1].kind = Kind::Dir(children);
204 }
205 ino
206 }
207
208 /// The listing of `ino` as built from the `.ls` (for `fetch`).
209 fn tree_of(&self, ino: u64) -> Tree {
210 match &self.nodes[ino as usize - 1].kind {
211 Kind::Dir(c) => Tree::Dir(
212 c.iter()
213 .map(|(n, i)| (n.to_string(), self.tree_of(*i as u64)))
214 .collect(),
215 ),
216 Kind::File { size, exec } => Tree::File {
217 size: *size,
218 exec: *exec,
219 },
220 Kind::Symlink(t) => Tree::Symlink(String::from_utf8_lossy(t).into_owned()),
221 }
222 }
223
224 /// Fetch store path `name` as an open() would and check that what got
225 /// unpacked matches its listing (`lazy-store fetch`, no FUSE needed).
226 pub fn fetch_and_check(&self, name: &str) -> io::Result<PathBuf> {
227 let c = CString::new(name).map_err(io::Error::other)?;
228 let ino = self.lookup_child(1, &c).ok_or_else(|| err(libc::ENOENT))?;
229 let p = self.fetch(self.nodes[ino as usize - 1].path as usize)?;
230 if Tree::from_path(&p)? != self.tree_of(ino) {
231 return Err(io::Error::other(format!(
232 "{name}: unpacked tree differs from its listing"
233 )));
234 }
235 Ok(p)
236 }
237
238 pub fn set_passthrough(&self, on: bool) {
239 self.passthrough
240 .store(on && self.backing.is_some(), Ordering::Relaxed);
241 }
242
243 fn node(&self, ino: u64) -> io::Result<&Node> {
244 (ino as usize)
245 .checked_sub(1)
246 .and_then(|i| self.nodes.get(i))
247 .ok_or_else(|| err(libc::ENOENT))
248 }
249
250 pub fn has_root_entry(&self, name: &CStr) -> bool {
251 self.lookup_child(1, name).is_some()
252 }
253
254 pub fn root_entries(&self) -> impl Iterator<Item = (&str, u64, u32)> {
255 let Kind::Dir(c) = &self.nodes[0].kind else {
256 unreachable!()
257 };
258 c.iter()
259 .map(|(n, i)| (&**n, *i as u64, self.dtype(*i as u64)))
260 }
261
262 fn lookup_child(&self, parent: u64, name: &CStr) -> Option<u64> {
263 let Kind::Dir(c) = &self.node(parent).ok()?.kind else {
264 return None;
265 };
266 let name = name.to_str().ok()?;
267 c.binary_search_by(|(n, _)| (**n).cmp(name))
268 .ok()
269 .map(|i| c[i].1 as u64)
270 }
271
272 fn dtype(&self, ino: u64) -> u32 {
273 match self.nodes[ino as usize - 1].kind {
274 Kind::Dir(_) => libc::DT_DIR as u32,
275 Kind::File { .. } => libc::DT_REG as u32,
276 Kind::Symlink(_) => libc::DT_LNK as u32,
277 }
278 }
279
280 fn stat(&self, ino: u64) -> io::Result<stat64> {
281 let n = self.node(ino)?;
282 // Safe: stat64 is plain old data.
283 let mut st: stat64 = unsafe { std::mem::zeroed() };
284 st.st_ino = ino | INO_TAG;
285 st.st_blksize = 4096;
286 st.st_atime = 1;
287 st.st_mtime = 1;
288 st.st_ctime = 1;
289 match &n.kind {
290 Kind::Dir(_) => {
291 st.st_mode = libc::S_IFDIR | 0o555;
292 // 1: "subdirectory count unknown", which keeps find(1)
293 // from skipping entries based on it
294 st.st_nlink = 1;
295 }
296 Kind::File { size, exec } => {
297 st.st_mode = libc::S_IFREG | if *exec { 0o555 } else { 0o444 };
298 st.st_nlink = 1;
299 st.st_size = *size as i64;
300 st.st_blocks = size.div_ceil(512) as i64;
301 }
302 Kind::Symlink(t) => {
303 st.st_mode = libc::S_IFLNK | 0o777;
304 st.st_nlink = 1;
305 st.st_size = t.len() as i64;
306 }
307 }
308 Ok(st)
309 }
310
311 /// The unpacked store path `i`, fetching it first if needed. Concurrent
312 /// callers for the same path wait for one fetch.
313 fn fetch(&self, i: usize) -> io::Result<PathBuf> {
314 let mut slot = self.fetched[i].lock().unwrap();
315 if let Some(p) = &*slot {
316 return Ok(p.clone());
317 }
318 let name = &self.names[i];
319 let dest = self.cache.join(name);
320 let mut last = None;
321 for attempt in 1..=3 {
322 match self.fetch_once(i, &dest) {
323 Ok(()) => {
324 *slot = Some(dest.clone());
325 return Ok(dest);
326 }
327 Err(e) => {
328 eprintln!("fetching {name} (attempt {attempt}/3): {e}");
329 last = Some(e);
330 }
331 }
332 }
333 Err(last.unwrap())
334 }
335
336 fn fetch_once(&self, i: usize, dest: &Path) -> io::Result<()> {
337 let (name, info) = (&self.names[i], &self.infos[i]);
338 let t = Instant::now();
339 let tmp = dest.with_extension("tmp");
340 nar::remove(&tmp)?;
341 nar::remove(dest)?;
342 let (body, _) = self.http.open(&format!("{}/{}", info.cache, info.url))?;
343 let mut counted = CountingReader { inner: body, n: 0 };
344 let (digest, size) = nar::unpack(http::decoder(&info.compression, &mut counted)?, &tmp)?;
345 if !nar::hash_matches(&info.nar_hash, &digest) || size != info.nar_size {
346 nar::remove(&tmp)?;
347 return Err(io::Error::other(format!(
348 "{name}: NAR hash mismatch, sha256:{} ({size} bytes) != {} ({} bytes)",
349 base32::encode_string(&digest),
350 info.nar_hash,
351 info.nar_size
352 )));
353 }
354 std::fs::rename(&tmp, dest)?;
355 self.stats.nar(name, counted.n, size);
356 eprintln!(
357 "fetched {name}: {} bytes in {:.1} s",
358 counted.n,
359 t.elapsed().as_secs_f64()
360 );
361 Ok(())
362 }
363
364 fn entry(&self, ino: u64) -> io::Result<Entry> {
365 Ok(Entry {
366 inode: ino,
367 generation: 0,
368 attr: self.stat(ino)?,
369 attr_flags: 0,
370 attr_timeout: TTL,
371 entry_timeout: TTL,
372 })
373 }
374
375 fn backing_id(&self, ino: u64, file: &File) -> Option<u32> {
376 if !self.passthrough.load(Ordering::Relaxed) {
377 return None;
378 }
379 let backing = self.backing.as_ref()?;
380 let mut ids = self.backing_ids.lock().unwrap();
381 if let Some((id, n)) = ids.get_mut(&ino) {
382 *n += 1;
383 return Some(*id);
384 }
385 match backing.open_fd(file.as_raw_fd()) {
386 Ok(id) => {
387 ids.insert(ino, (id, 1));
388 Some(id)
389 }
390 Err(e) => {
391 if !self.backing_failed.swap(true, Ordering::Relaxed) {
392 eprintln!(
393 "lower: FUSE passthrough backing open failed, reading via the daemon: {e}"
394 );
395 }
396 None
397 }
398 }
399 }
400
401 fn release_backing(&self, ino: u64) {
402 let mut ids = self.backing_ids.lock().unwrap();
403 if let Some((id, n)) = ids.get_mut(&ino) {
404 *n -= 1;
405 if *n == 0 {
406 let id = *id;
407 ids.remove(&ino);
408 drop(ids);
409 if let Some(b) = &self.backing {
410 b.close_id(id);
411 }
412 }
413 }
414 }
415}
416
417impl FileSystem for LazyFs {
418 type Inode = u64;
419 type Handle = u64;
420
421 fn lookup(&self, _ctx: &Context, parent: u64, name: &CStr) -> io::Result<Entry> {
422 match self.lookup_child(parent, name) {
423 Some(ino) => self.entry(ino),
424 None => Err(err(libc::ENOENT)),
425 }
426 }
427
428 fn getattr(&self, _ctx: &Context, ino: u64, _h: Option<u64>) -> io::Result<(stat64, Duration)> {
429 Ok((self.stat(ino)?, TTL))
430 }
431
432 fn readlink(&self, _ctx: &Context, ino: u64) -> io::Result<Vec<u8>> {
433 match &self.node(ino)?.kind {
434 Kind::Symlink(t) => Ok(t.to_vec()),
435 _ => Err(err(libc::EINVAL)),
436 }
437 }
438
439 fn open(
440 &self,
441 _ctx: &Context,
442 ino: u64,
443 flags: u32,
444 _fuse_flags: u32,
445 ) -> io::Result<(Option<u64>, OpenOptions, Option<u32>)> {
446 let n = self.node(ino)?;
447 if !matches!(n.kind, Kind::File { .. }) {
448 return Err(err(libc::EISDIR));
449 }
450 let acc = flags as i32 & libc::O_ACCMODE;
451 if acc != libc::O_RDONLY || flags as i32 & libc::O_TRUNC != 0 {
452 return Err(err(libc::EROFS));
453 }
454 let root = self.fetch(n.path as usize).map_err(|e| {
455 eprintln!("open {}/{}: {e}", self.names[n.path as usize], n.rel);
456 err(libc::EIO)
457 })?;
458 let file = File::open(if n.rel.is_empty() {
459 root
460 } else {
461 root.join(&*n.rel)
462 })?;
463 let backing = self.backing_id(ino, &file);
464 let h = self.next_handle.fetch_add(1, Ordering::Relaxed);
465 self.handles
466 .write()
467 .unwrap()
468 .insert(h, Arc::new(Handle { file, ino }));
469 Ok(match backing {
470 Some(id) => (Some(h), OpenOptions::PASSTHROUGH, Some(id)),
471 None => (Some(h), OpenOptions::KEEP_CACHE, None),
472 })
473 }
474
475 fn read(
476 &self,
477 _ctx: &Context,
478 _ino: u64,
479 handle: u64,
480 w: &mut dyn ZeroCopyWriter,
481 size: u32,
482 offset: u64,
483 _lock_owner: Option<u64>,
484 _flags: u32,
485 ) -> io::Result<usize> {
486 let h = self
487 .handles
488 .read()
489 .unwrap()
490 .get(&handle)
491 .cloned()
492 .ok_or_else(|| err(libc::EBADF))?;
493 // Safe: borrows the fd of `h.file`, which outlives `f`, without closing it.
494 let mut f = ManuallyDrop::new(unsafe { File::from_raw_fd(h.file.as_raw_fd()) });
495 w.write_from(&mut *f, size as usize, offset)
496 }
497
498 fn lseek(
499 &self,
500 _ctx: &Context,
501 _ino: u64,
502 handle: u64,
503 offset: u64,
504 whence: u32,
505 ) -> io::Result<u64> {
506 let h = self
507 .handles
508 .read()
509 .unwrap()
510 .get(&handle)
511 .cloned()
512 .ok_or_else(|| err(libc::EBADF))?;
513 // Safe: plain syscall on an fd we own.
514 let r = unsafe { libc::lseek(h.file.as_raw_fd(), offset as i64, whence as i32) };
515 if r < 0 {
516 return Err(io::Error::last_os_error());
517 }
518 Ok(r as u64)
519 }
520
521 fn flush(&self, _ctx: &Context, _ino: u64, _handle: u64, _lock_owner: u64) -> io::Result<()> {
522 Ok(())
523 }
524
525 fn fsync(&self, _ctx: &Context, _ino: u64, _datasync: bool, _handle: u64) -> io::Result<()> {
526 Ok(())
527 }
528
529 fn release(
530 &self,
531 _ctx: &Context,
532 _ino: u64,
533 _flags: u32,
534 handle: u64,
535 _flush: bool,
536 _flock_release: bool,
537 _lock_owner: Option<u64>,
538 ) -> io::Result<()> {
539 if let Some(h) = self.handles.write().unwrap().remove(&handle) {
540 self.release_backing(h.ino);
541 }
542 Ok(())
543 }
544
545 fn opendir(
546 &self,
547 _ctx: &Context,
548 ino: u64,
549 _flags: u32,
550 ) -> io::Result<(Option<u64>, OpenOptions)> {
551 match self.node(ino)?.kind {
552 Kind::Dir(_) => Ok((None, OpenOptions::KEEP_CACHE | OpenOptions::CACHE_DIR)),
553 _ => Err(err(libc::ENOTDIR)),
554 }
555 }
556
557 fn readdir(
558 &self,
559 _ctx: &Context,
560 ino: u64,
561 _handle: u64,
562 _size: u32,
563 offset: u64,
564 add_entry: &mut dyn FnMut(DirEntry) -> io::Result<usize>,
565 ) -> io::Result<()> {
566 let Kind::Dir(c) = &self.node(ino)?.kind else {
567 return Err(err(libc::ENOTDIR));
568 };
569 for (i, (name, child)) in c.iter().enumerate().skip(offset as usize) {
570 let child = *child as u64;
571 let n = add_entry(DirEntry {
572 ino: child | INO_TAG,
573 offset: i as u64 + 1,
574 type_: self.dtype(child),
575 name: name.as_bytes(),
576 })?;
577 if n == 0 {
578 break;
579 }
580 }
581 Ok(())
582 }
583
584 fn releasedir(&self, _ctx: &Context, _ino: u64, _flags: u32, _handle: u64) -> io::Result<()> {
585 Ok(())
586 }
587
588 fn getxattr(
589 &self,
590 _ctx: &Context,
591 _ino: u64,
592 _name: &CStr,
593 _size: u32,
594 ) -> io::Result<GetxattrReply> {
595 Err(err(libc::ENODATA))
596 }
597
598 fn listxattr(&self, _ctx: &Context, _ino: u64, size: u32) -> io::Result<ListxattrReply> {
599 Ok(if size == 0 {
600 ListxattrReply::Count(0)
601 } else {
602 ListxattrReply::Names(Vec::new())
603 })
604 }
605
606 fn access(&self, _ctx: &Context, _ino: u64, mask: u32) -> io::Result<()> {
607 if mask as i32 & libc::W_OK != 0 {
608 return Err(err(libc::EROFS));
609 }
610 Ok(())
611 }
612}