mesh-host/cmd/mesh-host/kept.go

package main

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/novox/mesh-host/internal/apply"
	"github.com/novox/mesh-host/internal/declaration"
	"github.com/novox/mesh-host/internal/link"
	"github.com/novox/mesh-host/internal/store"
)

// An access is judged by what each of its users may do (novox/hq ADR 0325 §7, §10, §12).
//
// On every look the engine judges each module's access as its module's resource of kind `access`: waiting, with a
// wait of kind access, while the mesh may not keep its folder yet or its disk is not there; unhealthy for root's
// data on its placed filesystem, and for a kept folder that gives a user less than its right — read by entry and
// mask together, on the folder and the sweep's sample under it, which the sweep puts back where it can; healthy
// otherwise. The folder is the target, never in a reason nor in a wait's words: the controller says them on the
// operator's channel, which withholds a path (ADR 0234 §6).

// judgedAccess is one access as the judge reads it.
type judgedAccess struct {
	module, id, name, path string
	taken                  bool
	users                  []store.KeptUser
	// leftover is the mesh's entries nobody wants, not an access (link.KindKeptEntries).
	leftover bool
}

// accessJudge reads every access on every look. Safe for the apply and the looking loop at once.
type accessJudge struct {
	mu       sync.Mutex
	accesses []judgedAccess
	// was is each access's last verdict, with since when and how many looks in a row found it unhealthy.
	was map[string]link.ResourceHealth

	// reading guards inFlight: every folder whose look has not returned yet.
	reading  sync.Mutex
	inFlight map[string]bool
}

// accesses is the one access judge on this machine.
var accesses = newAccessJudge()

func newAccessJudge() *accessJudge {
	return &accessJudge{was: map[string]link.ResourceHealth{}, inFlight: map[string]bool{}}
}

// set is what the declaration just applied asks: every access of a module, not held as found, with the users of
// its folder as every access of that folder gives them. An access no longer declared is forgotten.
func (j *accessJudge) set(d *declaration.Declaration, held map[string]bool) {
	var as []judgedAccess
	if d != nil {
		users := map[string]map[uint32]store.KeptUser{}
		taken := map[string]bool{}
		for _, r := range d.Resources {
			if a, ok := r.(*declaration.Access); ok {
				path := filepath.Clean(a.Path)
				if users[path] == nil {
					users[path] = map[uint32]store.KeptUser{}
				}
				taken[path] = taken[path] || a.Taken
				for _, u := range a.Users {
					had, seen := users[path][uint32(u.UID)]
					if !seen || (u.Use == declaration.AccessReadWrite && had.Use != declaration.AccessReadWrite) {
						users[path][uint32(u.UID)] = store.KeptUser{Identity: u.Identity, UID: uint32(u.UID),
							Use: u.Use, From: u.From}
					}
				}
			}
		}
		for _, r := range d.Resources {
			a, ok := r.(*declaration.Access)
			if !ok || held[a.ID] || strings.HasPrefix(a.ID, declaration.AdoptionPrefix) {
				continue
			}
			// An older controller's access is judged by nothing, as before ADR 0325: no state is said of it.
			if !apply.Keeps(d, a.Path) {
				continue
			}
			at := strings.LastIndex(a.ID, ".")
			if at <= 0 {
				continue
			}
			path := filepath.Clean(a.Path)
			ja := judgedAccess{module: a.ID[:at], id: a.ID, name: a.ID[at+1:], path: path, taken: taken[path]}
			for _, u := range users[path] {
				ja.users = append(ja.users, u)
			}
			sort.Slice(ja.users, func(x, y int) bool { return ja.users[x].UID < ja.users[y].UID })
			as = append(as, ja)
		}
	}
	sort.Slice(as, func(a, b int) bool { return as[a].id < as[b].id })
	j.mu.Lock()
	defer j.mu.Unlock()
	j.accesses = as
	declared := map[string]bool{}
	for _, a := range as {
		declared[a.id] = true
	}
	for id := range j.was {
		if !declared[id] {
			delete(j.was, id)
		}
	}
}

// lookBound is how long a look waits for one accessed folder to answer. A hard network mount whose server is down
// blocks a stat for ever, and one access must never stop every health statement this machine makes.
var lookBound = 5 * time.Second

// verdict is one access's judgement in a look.
type verdict struct {
	state, reason string
	waits         []link.AccessWait
	act           *link.AccessActRecord
	entries       []link.KeptEntry
}

// judgeBounded judges one access within lookBound, outside any lock. A look that has not answered is said so;
// while it still hangs, later looks start no other on the same folder, so a dead mount costs one blocked read.
func (j *accessJudge) judgeBounded(a judgedAccess) (verdict, bool) {
	j.reading.Lock()
	if j.inFlight[a.path] {
		j.reading.Unlock()
		return verdict{}, false
	}
	j.inFlight[a.path] = true
	j.reading.Unlock()
	done := make(chan verdict, 1)
	go func() {
		v := judgeAccess(a)
		j.reading.Lock()
		delete(j.inFlight, a.path)
		j.reading.Unlock()
		done <- v
	}()
	select {
	case v := <-done:
		return v, true
	case <-time.After(lookBound):
		return verdict{}, false
	}
}

// judgeAccess is one access's judgement now (ADR 0325 §7, §10, §12), the sweep of its kept folder included.
func judgeAccess(a judgedAccess) verdict {
	v := apply.AccessVerdict(context.Background(), a.path, a.taken, len(a.users) > 0)
	switch v.Kind {
	case apply.VerdictWaitKeep:
		return verdict{state: link.StateWaiting, reason: "it holds data the mesh has not taken on the filesystem " +
			"it is on now, and waits for the operator's word (novox/hq ADR 0325 §7, §8)",
			waits: []link.AccessWait{{Kind: link.WaitKindAccess, Access: a.name, Part: a.module,
				What: apply.KeepWords(a.name), Placed: v.Placed}}}
	case apply.VerdictWaitDisk:
		return verdict{state: link.StateWaiting, reason: "the disk that holds it is not there (novox/hq ADR 0325 §9)",
			waits: []link.AccessWait{{Kind: link.WaitKindAccess, Access: a.name, Part: a.module,
				What: apply.DiskWords(a.name)}}}
	case apply.VerdictRoot:
		return verdict{state: link.StateUnhealthy, reason: apply.RootWords(v.Why)}
	case apply.VerdictUnknown:
		// Not a fault and not the operator's: said as not known, and judged again on the next look.
		return verdict{state: link.StateUnknown, reason: v.Why + "; judged again on the next look"}
	case apply.VerdictBroken:
		// Its words carry no path (AccessVerdict says them so): the reason reaches the operator's channel.
		return verdict{state: link.StateUnhealthy, reason: v.Why}
	case apply.VerdictMake:
		return verdict{state: link.StateUnhealthy, reason: "it is not there yet; the next apply makes it where " +
			"its data lives (novox/hq ADR 0325 §9)"}
	case apply.VerdictNobody:
		return verdict{state: link.StateHealthy}
	}
	out := verdict{state: link.StateHealthy, entries: entriesOf(a.users)}
	rec, _ := keptNow(a.path)
	if act := rec.Act; act != nil {
		out.act = &link.AccessActRecord{Act: act.Act, Access: a.id, Warrant: act.Warrant, By: act.By, State: act.State,
			Started: act.Started, Ended: act.Ended, Folders: act.Folders, Files: act.Files, Kept: act.Kept,
			Links: act.Links, Other: act.Other, Mounts: act.Mounts, Why: act.Why}
	}
	if !rec.Kept {
		out.state, out.reason = link.StateStarting, "the controller says it was taken; the next apply keeps it"
		return out
	}
	look := apply.LookKept(a.path, a.users, apply.SweepLimits)
	var faults []string
	if look.Failed != "" {
		faults = append(faults, "the sweep could not look: "+apply.Pathless(errors.New(look.Failed)))
	}
	if len(look.Short) > 0 {
		faults = append(faults, "the folder itself gives "+strings.Join(look.Short, ", ")+"; put back on the next "+
			"apply")
	}
	if n := look.Sweep.Unrepaired + look.Sweep.LastUnrepaired; n > 0 {
		faults = append(faults, fmt.Sprintf("%d entries under it give a user less than its right and cannot be "+
			"put back: an owner's own bits, which the mesh never changes, or an entry the filesystem refused", n))
	}
	if len(faults) > 0 {
		out.state, out.reason = link.StateUnhealthy, strings.Join(faults, "; ")
		return out
	}
	switch {
	case rec.Act != nil && rec.Act.State == link.ActRunning:
		out.state, out.reason = link.StateStarting, "the take act is adding its users' entries under it"
	case !sameUsersList(rec.Users, rec.Want):
		out.state, out.reason = link.StateStarting, "the entries under it are being changed to its users now"
	}
	return out
}

// entriesOf is a kept folder's listing (ADR 0325 §4): each user, and beside the tool runner's what it stands for.
func entriesOf(users []store.KeptUser) []link.KeptEntry {
	var out []link.KeptEntry
	for _, u := range users {
		e := link.KeptEntry{Identity: u.Identity, UID: int(u.UID), Use: u.Use, From: u.From}
		if u.From == declaration.FromToolRunner {
			e.Note = "the tool runner's account stands for every module whose tools use this folder, at the widest " +
				"use any of them declares: a module that declared only read may write through it (novox/hq ADR " +
				"0325 §4, until issue 569)"
		}
		out = append(out, e)
	}
	return out
}

func sameUsersList(a, b []store.KeptUser) bool {
	if len(a) != len(b) {
		return false
	}
	seen := map[uint32]string{}
	for _, u := range a {
		seen[u.UID] = u.Use
	}
	for _, u := range b {
		if use, ok := seen[u.UID]; !ok || use != u.Use {
			return false
		}
	}
	return true
}

// keptNow is the record of one kept folder, as the engine's file holds it.
func keptNow(path string) (store.KeptFolder, bool) {
	all, err := apply.KeptFolders.All()
	if err != nil {
		return store.KeptFolder{}, false
	}
	f, had := all[path]
	return f, had
}

// look reads every access now. Whether any verdict changed since the last look, for the loop to say it.
func (j *accessJudge) look(now time.Time) ([]link.ResourceHealth, bool) {
	j.mu.Lock()
	as := append([]judgedAccess(nil), j.accesses...)
	j.mu.Unlock()

	read := make([]verdict, len(as))
	for i, a := range as {
		v, answered := j.judgeBounded(a)
		if !answered {
			v = verdict{state: link.StateUnhealthy, reason: "its folder does not answer: judging it did not return " +
				"within " + lookBound.String()}
		}
		read[i] = v
	}

	// And the entries nobody wants that could not be taken out yet, each under the access that last named its
	// folder, until its drop walk has run (ADR 0325 §8).
	for _, l := range apply.Leftovers() {
		as = append(as, judgedAccess{module: l.Module, id: l.Access + "~entries", name: l.Name, path: l.Path,
			leftover: true})
		read = append(read, verdict{state: link.StateUnhealthy, reason: l.Why})
	}

	j.mu.Lock()
	defer j.mu.Unlock()
	changed := false
	out := make([]link.ResourceHealth, 0, len(as))
	for i, a := range as {
		r := read[i]
		kind := link.KindAccess
		if a.leftover {
			kind = link.KindKeptEntries
		}
		v := link.ResourceHealth{Module: a.module, Resource: a.id, Kind: kind, Target: a.path,
			State: r.state, Reason: r.reason, Since: now.UTC(), AccessWaits: r.waits, Act: r.act, Entries: r.entries}
		if was, had := j.was[a.id]; had && was.State == r.state && was.Reason == r.reason {
			v.Since = was.Since
			v.Streak = was.Streak
		} else {
			changed = true
		}
		if r.state == link.StateUnhealthy {
			v.Streak++
		}
		j.was[a.id] = v
		out = append(out, v)
	}
	return out, changed
}

// withAccesses adds to a statement every access, as its module's resource of kind access.
func withAccesses(h *link.Health, vs []link.ResourceHealth) *link.Health {
	h.Resources = append(h.Resources, vs...)
	return h
}

// accessesHealthy says no access in vs is anything but healthy.
func accessesHealthy(vs []link.ResourceHealth) bool {
	for _, v := range vs {
		if v.State != link.StateHealthy {
			return false
		}
	}
	return true
}

// keeper runs the walks under kept folders beside the applies, one at a time (ADR 0325 §8): poked after every
// apply and every look, it walks whatever folder's entries under it are not what was declared.
type keeper struct {
	running, following atomic.Bool
	say                func(string)
	watch              *apply.Watcher
}

// keeping is the one keeper on this machine.
var keeping = &keeper{say: func(string) {}}

// poke starts a reconcile of the kept folders unless one is running, and has the watch follow what is kept.
func (k *keeper) poke(ctx context.Context) {
	// The watch is set beside, never in the caller's way: watching a new folder's tree reads every folder of it.
	if k.following.CompareAndSwap(false, true) {
		go func() {
			defer k.following.Store(false)
			k.follow()
		}()
	}
	if !k.running.CompareAndSwap(false, true) {
		return
	}
	go func() {
		defer k.running.Store(false)
		for _, line := range apply.ReconcileKept(ctx) {
			k.say(line)
		}
	}()
}

// follow has the watch watch every kept folder with its users.
func (k *keeper) follow() {
	if k.watch == nil {
		return
	}
	all, err := apply.KeptFolders.All()
	if err != nil {
		return
	}
	folders := map[string][]store.KeptUser{}
	for path, f := range all {
		if f.Kept && len(f.Want) > 0 {
			folders[path] = f.Want
		}
	}
	k.watch.Set(folders)
}

// startKeeping opens the watch and has the keeper say what it does. A machine without the watch is said so, and
// its sweep alone finds what is made.
func startKeeping(ctx context.Context, say func(string)) {
	keeping.say = say
	w, err := apply.NewWatcher()
	if err != nil {
		say(fmt.Sprintf("the kept folders are not watched (%v): the sweep alone finds what is made in them", err))
	} else {
		keeping.watch = w
		go w.Run(ctx)
	}
	keeping.poke(ctx)
}

// accessActFrom is the take act as the controller asks it over the link (ADR 0325 §8): judged against the
// declaration this node keeps, and run beside the applies.
func accessActFrom(statePath string, timeout time.Duration) link.AccessActFunc {
	return func(act link.AccessAct) (string, error) {
		kept, err := store.ReadDeclared(store.DeclaredPath(statePath))
		if errors.Is(err, store.ErrNothingDeclared) {
			return "", errors.New("this node keeps no declaration, so it is declared no access")
		}
		if err != nil {
			return "", err
		}
		d, err := declaration.Parse(kept.Declaration)
		if err != nil {
			return "", fmt.Errorf("the declaration this node keeps does not read: %w", err)
		}
		ctx, cancel := context.WithTimeout(context.Background(), timeout)
		defer cancel()
		_, said, err := apply.TakeFolder(ctx, d, act.Access, filepath.Clean(act.Path), act.Warrant, act.By,
			apply.ExecRunner)
		if err != nil {
			fmt.Fprintf(os.Stderr, "mesh-host: the take act on %s was refused: %v\n", act.Access, err)
			return "", err
		}
		return said, nil
	}
}

// withWaiting is held with every id of waiting beside it: what liveness does not judge.
func withWaiting(held, waiting map[string]bool) map[string]bool {
	if len(waiting) == 0 {
		return held
	}
	out := map[string]bool{}
	for id := range held {
		out[id] = true
	}
	for id := range waiting {
		out[id] = true
	}
	return out
}