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