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