//go:build linux package apply import ( "context" "errors" "strconv" "sync" "time" "unsafe" "golang.org/x/sys/unix" "github.com/novox/mesh-host/internal/store" ) // The watch on each kept folder (novox/hq ADR 0325 §6). // // What an identity makes in a kept folder inherits the default entries, and its mask follows the mode its maker // asked for: a file made at 0644 gives a read-write user only read until its mask is widened. The engine watches // every folder of each kept tree (inotify: made, moved in, attributes changed) and, on notice, gives the entry its // users' entries and a mask that covers them — the same keepVisit the walks use, through the entry's own // descriptor, reached from the kept folder down by O_NOFOLLOW, never through a link and never on another // filesystem. Notice can be lost (the kernel's queue overflowed) and watches run out (the machine's limit, ENOSPC): // then the record says so and the sweep finds what was made where no watch is (§10). // watchMask is what a watch hears: an entry made or moved in, and an entry's mode or ACL changed. // It is added through /proc/self/fd, whose link is the folder's own descriptor: following it is what reaches the // folder that was opened, so the watch takes no IN_DONT_FOLLOW. const watchMask = unix.IN_CREATE | unix.IN_MOVED_TO | unix.IN_ATTRIB | unix.IN_ONLYDIR // MaxWatchedPerFolder bounds the folders watched under one kept folder: beyond it the sweep alone keeps the rest. var MaxWatchedPerFolder = 100_000 // Watcher watches every kept folder's tree. type Watcher struct { mu sync.Mutex fd int dirs map[int]watchedDir roots map[string]*watchedRoot } // watchedDir is one watched folder: its kept folder and the names below it. type watchedDir struct { root string rel []string } // watchedRoot is one kept folder watched: its users, its device, and its watches. type watchedRoot struct { users map[uint32]bool dev uint64 wds []int out bool } // NewWatcher opens the machine's notice of changes. func NewWatcher() (*Watcher, error) { // Not blocking: Run polls it, so the end of its context ends it whatever the kernel has to say. fd, err := unix.InotifyInit1(unix.IN_CLOEXEC | unix.IN_NONBLOCK) if err != nil { return nil, err } return &Watcher{fd: fd, dirs: map[int]watchedDir{}, roots: map[string]*watchedRoot{}}, nil } // Set is every kept folder to watch, with its users: one newly kept is watched whole (within // MaxWatchedPerFolder), one no longer kept is let go, and one whose users changed is repaired with the new ones. func (w *Watcher) Set(folders map[string][]store.KeptUser) { w.mu.Lock() defer w.mu.Unlock() for path, root := range w.roots { if _, still := folders[path]; !still { for _, wd := range root.wds { _, _ = unix.InotifyRmWatch(w.fd, uint32(wd)) delete(w.dirs, wd) } delete(w.roots, path) } } for path, users := range folders { if root, had := w.roots[path]; had { root.users = usersOf(users) continue } root := &watchedRoot{users: usersOf(users)} w.roots[path] = root w.watchTree(path, root) w.say(path, root) } } // say records how many folders of a kept folder are watched, and whether the watches ran out. func (w *Watcher) say(path string, root *watchedRoot) { n, out := len(root.wds), root.out _ = KeptFolders.Update(path, func(f *store.KeptFolder, had bool) bool { f.Watches, f.OutOfWatches = n, out return had }) } // watchTree watches a kept folder and every folder under it on its device, by descriptor: each watch is added // through /proc/self/fd of the folder's own descriptor, so no name is followed to another place. func (w *Watcher) watchTree(path string, root *watchedRoot) { _, fd, st, err := openKept(path, WalkLimits{Entries: 1, Time: time.Minute}, machineStats, nil) if err != nil { return } root.dev = uint64(st.Dev) w.watchBelow(path, root, fd, nil) } // watchBelow watches the folder open on fd, rel below root, and every folder under it. It closes fd. func (w *Watcher) watchBelow(path string, root *watchedRoot, fd int, rel []string) { defer unix.Close(fd) if root.out || len(root.wds) >= MaxWatchedPerFolder { root.out = true return } wd, err := unix.InotifyAddWatch(w.fd, "/proc/self/fd/"+strconv.Itoa(fd), watchMask) if errors.Is(err, unix.ENOSPC) { root.out = true return } if err != nil { return } root.wds = append(root.wds, wd) w.dirs[wd] = watchedDir{root: path, rel: append([]string(nil), rel...)} dup, err := unix.Dup(fd) if err != nil { return } names, err := readNames(dup) if err != nil { return } for _, name := range names { var st unix.Stat_t if unix.Fstatat(fd, name, &st, unix.AT_SYMLINK_NOFOLLOW) != nil || st.Mode&unix.S_IFMT != unix.S_IFDIR || uint64(st.Dev) != root.dev { continue } next, err := unix.Openat(fd, name, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0) if err != nil { continue } w.watchBelow(path, root, next, append(rel, name)) } } // watchPoll is how long Run waits for notice before it looks whether its context ended. const watchPoll = 250 * time.Millisecond // Run reads notices until ctx ends, repairing each entry made or changed in a watched folder, and then closes the // watch: polled, so a quiet machine's watch ends as soon as a busy one's. func (w *Watcher) Run(ctx context.Context) { defer unix.Close(w.fd) buf := make([]byte, 64*1024) for ctx.Err() == nil { ready, err := unix.Poll([]unix.PollFd{{Fd: int32(w.fd), Events: unix.POLLIN}}, int(watchPoll/time.Millisecond)) if err != nil && !errors.Is(err, unix.EINTR) { return } if ready <= 0 { continue } n, err := unix.Read(w.fd, buf) if errors.Is(err, unix.EAGAIN) || errors.Is(err, unix.EINTR) { continue } if err != nil || n <= 0 { return } for at := 0; at+unix.SizeofInotifyEvent <= n; { ev := (*unix.InotifyEvent)(unsafe.Pointer(&buf[at])) nameAt := at + unix.SizeofInotifyEvent name := cString(buf[nameAt : nameAt+int(ev.Len)]) at = nameAt + int(ev.Len) if ev.Mask&unix.IN_Q_OVERFLOW != 0 { w.lost() continue } if ev.Mask&unix.IN_IGNORED != 0 { // The watch is gone — its folder removed, or let go by Set: what pointed at it goes too. w.forget(int(ev.Wd)) continue } w.heard(int(ev.Wd), name, ev.Mask) } } } // lost is notice lost (the kernel's queue overflowed): every kept folder is watched afresh, and what was made // meanwhile is the sweep's. func (w *Watcher) lost() { w.mu.Lock() defer w.mu.Unlock() for path, root := range w.roots { for _, wd := range root.wds { _, _ = unix.InotifyRmWatch(w.fd, uint32(wd)) delete(w.dirs, wd) } root.wds, root.out = nil, false w.watchTree(path, root) w.say(path, root) } } // forget lets go of a watch the kernel removed. func (w *Watcher) forget(wd int) { w.mu.Lock() defer w.mu.Unlock() dir, ok := w.dirs[wd] if !ok { return } delete(w.dirs, wd) if root := w.roots[dir.root]; root != nil { for i, had := range root.wds { if had == wd { root.wds = append(root.wds[:i], root.wds[i+1:]...) break } } w.say(dir.root, root) } } // watched is how many folders are watched now. func (w *Watcher) watched() int { w.mu.Lock() defer w.mu.Unlock() return len(w.dirs) } // heard repairs one entry a watch heard of: name in the watched folder wd, or the folder itself. func (w *Watcher) heard(wd int, name string, mask uint32) { w.mu.Lock() dir, ok := w.dirs[wd] root := w.roots[dir.root] var users map[uint32]bool var dev uint64 if root != nil { // Read under the lock: Set gives a root new users while notices are heard. users, dev = root.users, root.dev } w.mu.Unlock() if !ok || root == nil { return } fd, err := openBelow(dir.root, dir.rel, dev) if err != nil { return } defer unix.Close(fd) visit := keepVisit(users) if name == "" { var st unix.Stat_t if unix.Fstat(fd, &st) == nil { _, _, _, _ = visit(fd, st, true) } return } var st unix.Stat_t if unix.Fstatat(fd, name, &st, unix.AT_SYMLINK_NOFOLLOW) != nil || uint64(st.Dev) != dev { return } isDir := st.Mode&unix.S_IFMT == unix.S_IFDIR if !isDir && st.Mode&unix.S_IFMT != unix.S_IFREG { return } flags := unix.O_RDONLY | unix.O_NOFOLLOW | unix.O_CLOEXEC | unix.O_NOCTTY | unix.O_NONBLOCK if isDir { flags |= unix.O_DIRECTORY } efd, err := unix.Openat(fd, name, flags, 0) if err != nil { return } var opened unix.Stat_t if unix.Fstat(efd, &opened) != nil || opened.Ino != st.Ino || opened.Dev != st.Dev { unix.Close(efd) return } _, _, _, _ = visit(efd, opened, isDir) if isDir && mask&(unix.IN_CREATE|unix.IN_MOVED_TO) != 0 { // A folder made or moved in is watched, and what was moved in with it is the sweep's. w.mu.Lock() defer w.mu.Unlock() if w.roots[dir.root] != root { // Let go by Set meanwhile: nothing more is watched for it. unix.Close(efd) return } w.watchBelow(dir.root, root, efd, append(append([]string(nil), dir.rel...), name)) w.say(dir.root, root) return } unix.Close(efd) } // openBelow opens the folder rel below the kept folder root, each level by O_NOFOLLOW on root's device. func openBelow(root string, rel []string, dev uint64) (int, error) { _, fd, _, err := openKept(root, WalkLimits{Entries: 1, Time: time.Minute}, machineStats, nil) if err != nil { return -1, err } for _, name := range rel { next, err := unix.Openat(fd, name, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0) unix.Close(fd) if err != nil { return -1, err } var st unix.Stat_t if err := unix.Fstat(next, &st); err != nil || uint64(st.Dev) != dev { unix.Close(next) return -1, errors.New("a level is on another filesystem") } fd = next } return fd, nil } // readNames reads every name in the folder open on fd, and closes it. func readNames(fd int) ([]string, error) { var names []string buf := make([]byte, 32*1024) defer unix.Close(fd) for { n, err := unix.ReadDirent(fd, buf) if err != nil { return names, err } if n <= 0 { return names, nil } _, _, names = unix.ParseDirent(buf[:n], -1, names) } } func cString(b []byte) string { for i, c := range b { if c == 0 { return string(b[:i]) } } return string(b) }