mesh-delivery/modules/mesh-delivery-sdk/steprun/watch_settings_linux.go

//go:build linux

package steprun

import (
	"context"
	"encoding/binary"
	"os"
	"path/filepath"
	"syscall"
)

const settingsMask = syscall.IN_CLOSE_WRITE | syscall.IN_MOVED_TO | syscall.IN_CREATE

// watchSettings wakes the puller each time its settings file is written or replaced (inotify on the file's directory, so
// a rename over it counts): the event the node-engine's refill of the file is, which a parked puller at capacity 0
// otherwise never sees. No timer: it blocks on the kernel's event. When the kernel drops the watch (IN_IGNORED: the
// directory was replaced or removed) it is added again, said once; while the directory is missing the nearest existing
// ancestor is watched for the directory's return. A watch that cannot start is said once.
func (p *Puller) watchSettings(ctx context.Context) {
	if p.SettingsFile == "" {
		return
	}
	fd, err := syscall.InotifyInit1(syscall.IN_CLOEXEC | syscall.IN_NONBLOCK)
	if err != nil {
		p.logf("the settings file %s cannot be watched, so a refill of it wakes the puller only at another event: %v", p.SettingsFile, err)
		return
	}
	f := os.NewFile(uintptr(fd), "inotify")
	defer f.Close()
	dir, base := filepath.Split(p.SettingsFile)
	dir = filepath.Clean(dir)
	// arm watches dir, or when it is missing its nearest existing ancestor; it returns the watch descriptor and whether it
	// is the directory's own.
	arm := func() (int, bool, error) {
		wd, err := syscall.InotifyAddWatch(fd, dir, settingsMask)
		if err == nil {
			return wd, true, nil
		}
		for up := filepath.Dir(dir); ; up = filepath.Dir(up) {
			wd, aerr := syscall.InotifyAddWatch(fd, up, settingsMask)
			if aerr == nil {
				// the directory may have come back between the two calls: look once more, now that the ancestor is watched
				if own, oerr := syscall.InotifyAddWatch(fd, dir, settingsMask); oerr == nil {
					return own, true, nil
				}
				return wd, false, nil
			}
			if up == filepath.Dir(up) {
				return 0, false, err
			}
		}
	}
	wd, own, err := arm()
	if err != nil {
		p.logf("the settings file %s cannot be watched, so a refill of it wakes the puller only at another event: %v", p.SettingsFile, err)
		return
	}
	if p.watching != nil {
		p.watching()
	}
	go func() { <-ctx.Done(); f.Close() }()
	buf := make([]byte, 4096)
	for {
		n, err := f.Read(buf)
		if err != nil {
			return
		}
		hit, rearm := false, false
		for off := 0; off+16 <= n; {
			ewd := int(int32(binary.LittleEndian.Uint32(buf[off:])))
			mask := binary.LittleEndian.Uint32(buf[off+4:])
			l := int(binary.LittleEndian.Uint32(buf[off+12:]))
			name := buf[off+16 : off+16+l]
			for i, c := range name {
				if c == 0 {
					name = name[:i]
					break
				}
			}
			switch {
			case mask&syscall.IN_IGNORED != 0 && ewd == wd:
				rearm = true
			case ewd == wd && own && string(name) == base:
				hit = true
			case ewd == wd && !own:
				rearm = true // something appeared where the directory was: look again
			}
			off += 16 + l
		}
		if rearm {
			nwd, nown, err := arm()
			if err != nil {
				p.logf("the settings file %s is no longer watched, so a refill of it wakes the puller only at another event: %v", p.SettingsFile, err)
				return
			}
			if own || nown { // said once per loss of the watch on the directory itself
				if own {
					p.logf("the watch of the settings file %s was dropped (its directory was replaced); watching again", p.SettingsFile)
				}
			}
			wd, own = nwd, nown
			if p.watching != nil {
				p.watching()
			}
			hit = true // the file may have been refilled meanwhile
		}
		if hit {
			p.Wake()
		}
	}
}