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) {
	fd, err := unix.InotifyInit1(unix.IN_CLOEXEC)
	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))
	}
}

// Run reads notices until ctx ends, repairing each entry made or changed in a watched folder.
func (w *Watcher) Run(ctx context.Context) {
	go func() {
		<-ctx.Done()
		unix.Close(w.fd)
	}()
	buf := make([]byte, 64*1024)
	for {
		n, err := unix.Read(w.fd, buf)
		if err != nil || n <= 0 {
			if ctx.Err() != nil || !errors.Is(err, unix.EINTR) {
				return
			}
			continue
		}
		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
			}
			w.heard(int(ev.Wd), name, ev.Mask)
		}
	}
}

// lost is notice lost: every kept folder is said to rely on its sweep until it is watched afresh.
func (w *Watcher) lost() {
	w.mu.Lock()
	defer w.mu.Unlock()
	for path, root := range w.roots {
		root.out = true
		w.say(path, root)
	}
}

// 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]
	w.mu.Unlock()
	if !ok || root == nil {
		return
	}
	fd, err := openBelow(dir.root, dir.rel, root.dev)
	if err != nil {
		return
	}
	defer unix.Close(fd)
	visit := keepVisit(root.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) != root.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()
		w.watchBelow(dir.root, root, efd, append(append([]string(nil), dir.rel...), name))
		w.say(dir.root, root)
		w.mu.Unlock()
		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)
}