mesh-host/internal/apply/watch_linux.go

//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)
}