mesh-host/cmd/mesh-host/libraries.go

package main

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"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"
)

// A library is judged on every look (novox/hq ADR 0328 §8, §9).
//
// Each library this machine is reached by is said as a resource `library:<name>` of kind library, with no module:
// healthy while it is kept (or, off its home, while its mount is there), waiting with an access wait of reason keep
// or disk, unknown while its filesystem could not be named, unhealthy otherwise. The root 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). An access is judged by nothing, as before ADR 0325.

// libraryJudge reads every library on every look. Safe for the apply and the looking loop at once.
type libraryJudge struct {
	mu        sync.Mutex
	libraries []declaration.Library
	declared  *declaration.Declaration
	// was is each library'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 library whose look has not returned yet.
	reading  sync.Mutex
	inFlight map[string]bool
}

// accesses is the one library judge on this machine (its name from when it judged accesses).
var accesses = newLibraryJudge()

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

// set is what the declaration just applied declares: every library. One no longer declared is forgotten.
func (j *libraryJudge) set(d *declaration.Declaration, _ map[string]bool) {
	var ls []declaration.Library
	if d != nil {
		ls = append(ls, d.Libraries...)
	}
	sort.Slice(ls, func(a, b int) bool { return ls[a].Library < ls[b].Library })
	j.mu.Lock()
	defer j.mu.Unlock()
	j.libraries, j.declared = ls, d
	declared := map[string]bool{}
	for _, l := range ls {
		declared[apply.LibraryAccess(l.Library)] = true
	}
	for id := range j.was {
		if !declared[id] {
			delete(j.was, id)
		}
	}
}

// declaration is the declaration last set.
func (j *libraryJudge) declaration() *declaration.Declaration {
	j.mu.Lock()
	defer j.mu.Unlock()
	return j.declared
}

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

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

// judgeBounded judges one library 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 library.
func (j *libraryJudge) judgeBounded(l declaration.Library) (verdict, bool) {
	j.reading.Lock()
	if j.inFlight[l.Library] {
		j.reading.Unlock()
		return verdict{}, false
	}
	j.inFlight[l.Library] = true
	j.reading.Unlock()
	done := make(chan verdict, 1)
	go func() {
		v := judgeLibrary(l)
		j.reading.Lock()
		delete(j.inFlight, l.Library)
		j.reading.Unlock()
		done <- v
	}()
	select {
	case v := <-done:
		return v, true
	case <-time.After(lookBound):
		return verdict{}, false
	}
}

// judgeLibrary is one library's judgement now (ADR 0328 §7, §8, §9).
func judgeLibrary(l declaration.Library) verdict {
	v := apply.JudgeLibrary(context.Background(), l)
	access := apply.LibraryAccess(l.Library)
	var out verdict
	if a := v.Act; a != nil {
		out.act = &link.AccessActRecord{Act: a.Act, Access: access, Warrant: a.Warrant, By: a.By, State: a.State,
			Started: a.Started, Ended: a.Ended, Folders: a.Folders, Files: a.Files, Kept: a.Kept, Links: a.Links,
			Other: a.Other, Mounts: a.Mounts, Shared: a.Shared, Linked: a.Linked, Stripped: a.Stripped, Why: a.Why}
	}
	switch v.Kind {
	case apply.VerdictWaitKeep:
		out.state, out.reason = link.StateWaiting, "its root holds data the mesh has not taken on the filesystem "+
			"it is on now, and waits for the operator's word (novox/hq ADR 0328 §7, §8)"
		out.waits = []link.AccessWait{{Kind: link.WaitKindAccess, Access: access, Part: l.Library,
			What: apply.KeepWords(l.Library), Placed: v.Placed, Reason: link.WaitKeep}}
	case apply.VerdictWaitDisk:
		out.state, out.reason = link.StateWaiting, v.Why+" (novox/hq ADR 0328 §9)"
		out.waits = []link.AccessWait{{Kind: link.WaitKindAccess, Access: access, Part: l.Library,
			What: apply.DiskWords(l.Library), Reason: link.WaitDisk}}
	case apply.VerdictUnknown:
		// Not a fault and not the operator's: said as not known, and judged again on the next look.
		out.state, out.reason = link.StateUnknown, v.Why+"; judged again on the next look"
	case apply.VerdictBroken:
		out.state, out.reason = link.StateUnhealthy, v.Why
	default:
		out.state = link.StateHealthy
		switch {
		case v.Act != nil && v.Act.State == link.ActRunning:
			out.state, out.reason = link.StateStarting, "the take act is setting its group under its root"
		case v.Act != nil && (v.Act.State == link.ActStopped || v.Act.State == link.ActFailed):
			out.state, out.reason = link.StateStarting, "the take act goes on from where it stopped: "+v.Act.Why
		}
	}
	return out
}

// look reads every library now. Whether any verdict changed since the last look, for the loop to say it.
func (j *libraryJudge) look(now time.Time) ([]link.ResourceHealth, bool) {
	j.mu.Lock()
	ls := append([]declaration.Library(nil), j.libraries...)
	j.mu.Unlock()

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

	// And every tool runner's account the last apply could not put in a group, as a resource of its own (ADR 0334
	// §2): unhealthy, never failing the apply.
	warnings := apply.RunnerWarnings()

	j.mu.Lock()
	defer j.mu.Unlock()
	changed := false
	out := make([]link.ResourceHealth, 0, len(ls)+len(warnings))
	for _, w := range warnings {
		v := link.ResourceHealth{Module: w.Module, Resource: w.Resource, Kind: link.KindRunner, State: link.StateUnhealthy,
			Reason: "the tool runner's account joins no group: " + w.Why, Since: now.UTC(), Streak: 1}
		if was, had := j.was[w.Resource]; had && was.Reason == v.Reason {
			v.Since, v.Streak = was.Since, was.Streak+1
		} else {
			changed = true
		}
		j.was[w.Resource] = v
		out = append(out, v)
	}
	for i, l := range ls {
		r := read[i]
		id := apply.LibraryAccess(l.Library)
		v := link.ResourceHealth{Resource: id, Kind: link.KindLibrary, Target: l.Root, State: r.state,
			Reason: r.reason, Since: now.UTC(), AccessWaits: r.waits, Act: r.act}
		if was, had := j.was[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[id] = v
		out = append(out, v)
	}
	return out, changed
}

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

// accessesHealthy says no library 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 goes on with take acts stopped at their bound, beside the applies, one at a time (ADR 0328 §7).
type keeper struct {
	running atomic.Bool
	say     func(string)
}

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

// poke goes on with every stopped take act of the declaration last applied, unless that is running already.
func (k *keeper) poke(ctx context.Context) {
	d := accesses.declaration()
	if d == nil || len(d.Libraries) == 0 || !k.running.CompareAndSwap(false, true) {
		return
	}
	go func() {
		defer k.running.Store(false)
		for _, line := range apply.ReconcileLibraries(ctx, d) {
			k.say(line)
		}
	}()
}

// startKeeping has the keeper say what it does.
func startKeeping(ctx context.Context, say func(string)) {
	keeping.say = say
	keeping.poke(ctx)
}

// accessActFrom is the take act as the controller asks it over the link (ADR 0328 §7): 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 library")
		}
		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.TakeLibrary(ctx, d, act.Access, filepath.Clean(act.Path), act.Warrant, act.By)
		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
}