mesh-host/cmd/mesh-host/engineseat.go

package main

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync/atomic"
	"syscall"
	"time"

	nec "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
	occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient"
	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
	stdio "git.novox.be/novox/mesh-sdk/go"
	"github.com/novox/mesh-host/internal/accounts"
	"github.com/novox/mesh-host/internal/apply"
	"github.com/novox/mesh-host/internal/link"
	"github.com/novox/mesh-host/internal/selfcompose"
	"github.com/novox/mesh-host/internal/store"
)

// The node seat node-engine's judgements on this machine (novox/hq ADR 0563): what a hand-over and a library's take
// would set and what the disk holds now, and whether this machine is root-free.

// diskOf is what is on disk at a directory, as an act compares it when the operator's yes arrives: its owner and
// group, its mode, and which file it is.
func diskOf(info os.FileInfo) string {
	owner := "unknown"
	if uid, gid, ok := apply.OwnerOf(info); ok {
		owner = fmt.Sprintf("%d:%d", uid, gid)
	}
	which := ""
	if st, ok := info.Sys().(*syscall.Stat_t); ok {
		which = fmt.Sprintf(" %v:%v", st.Dev, st.Ino)
	}
	return fmt.Sprintf("%s %o%s", owner, info.Mode().Perm(), which)
}

// handOverJudgeFrom judges a hand-over through the node seat as the controller's ask is judged (judgeFound): the
// declared owner and mode it will set, and the disk as it is.
func handOverJudgeFrom(statePath string) func(string) (link.Judged, error) {
	return func(path string) (link.Judged, error) {
		found, info, err := judgeFound(statePath, path)
		if err != nil {
			return link.Judged{}, err
		}
		return link.Judged{Subject: path, Path: path, Sets: orUnsaid(found.Declared), Disk: diskOf(info),
			Words: "A directory of " + moduleOf(found.ID) + " is used as found: the mesh left its owner and mode."}, nil
	}
}

func moduleOf(id string) string {
	if at := strings.LastIndex(id, "."); at > 0 {
		return id[:at]
	}
	return id
}

// accessJudgeFrom judges the take act on a library through the node seat (ADR 0328 §7, ADR 0563 §4): declared on this
// machine, its home, and its root holding data the mesh has not taken; what the take sets, and the disk as it is.
func accessJudgeFrom(statePath string) func(string, string) (link.Judged, error) {
	return func(act, access string) (link.Judged, error) {
		d, err := declaredNow(statePath, nil)
		if errors.Is(err, store.ErrNothingDeclared) {
			return link.Judged{}, errors.New("this node keeps no declaration, so it is declared no library")
		}
		if err != nil {
			return link.Judged{}, fmt.Errorf("what runs on this node does not read: %w", err)
		}
		for _, l := range d.Libraries {
			if apply.LibraryAccess(l.Library) != access {
				continue
			}
			if !l.Home {
				return link.Judged{}, fmt.Errorf("this machine is not the home of the library %s, and a library is "+
					"taken on its home", l.Library)
			}
			ctx, cancel := context.WithTimeout(context.Background(), lookBound)
			defer cancel()
			if v := apply.JudgeLibrary(ctx, l); v.Kind != apply.VerdictWaitKeep {
				return link.Judged{}, fmt.Errorf("the library %s does not wait for the operator's word to be kept "+
					"(it is judged %s), so there is nothing to take", l.Library, v.Kind)
			}
			info, err := os.Lstat(l.Root)
			if err != nil {
				return link.Judged{}, fmt.Errorf("the library %s's root cannot be read: %v", l.Library, err)
			}
			return link.Judged{Subject: access, Path: filepath.Clean(l.Root),
				Sets: "the group " + l.Group + ", group write and setgid on everything under its root",
				Disk: diskOf(info), Words: "The library " + l.Library + " holds data the mesh has not taken."}, nil
		}
		return link.Judged{}, fmt.Errorf("the access %s names no library declared on this machine", access)
	}
}

// askTakes asks the take of every library the look found waiting for the operator's word, once per library (ADR 0563
// §4): asked by the engine itself, never a second time while one is open or after one was asked; access-act asks again
// by hand.
func askTakes(acts *link.HandActs, waiting []string, say func(string)) {
	if acts == nil {
		return
	}
	for _, access := range waiting {
		if acts.AskedFor(link.ActAccess, access) {
			continue
		}
		if _, err := acts.AskAccess(link.EngineSeat+"-"+acts.Node, link.ActTake, access,
			"its root holds data outside its group", true); err != nil && say != nil {
			say("the take of " + access + " could not be asked: " + err.Error())
		}
	}
}

// lastAccounts is the newest look at every account (the health loop's), which root-free judges from.
var lastAccounts atomic.Pointer[accounts.Statement]

// rootFreshFor is how old the verdict root-free judges from may be (ADR 0563 §6, the controller's verdictFreshFor).
const rootFreshFor = 15 * time.Minute

// judgeRootFree is this machine's root-free (ADR 0259 §8, ADR 0563 §6): free only on the engine's own verdict, at most
// rootFreshFor old, that every account declared never to become root without a person cannot, at least one such
// account declared, and the login shell's execute not served here. Anything else is no pass, and says why.
func judgeRootFree(node string, st *accounts.Statement, now time.Time, served bool, servedWhy string) link.RootVerdict {
	v := link.RootVerdict{Machine: node, Judged: now.UTC()}
	var not []string
	confined := ""
	switch {
	case st == nil:
		not = append(not, "this machine's engine has judged no account yet")
	case now.Sub(st.At) > rootFreshFor:
		not = append(not, fmt.Sprintf("the engine's newest judgement of its accounts was made at %s, more than %d "+
			"minutes ago, and a verdict that old is not a verdict on now", st.At.UTC().Format(time.RFC3339),
			int(rootFreshFor.Minutes())))
	default:
		var named []string
		for _, a := range st.Accounts {
			if !a.Root {
				continue
			}
			named = append(named, a.Name)
			if a.State != accounts.Healthy {
				not = append(not, "the account "+a.Name+" "+orNoneKnownWord(a.Reason))
			}
		}
		if len(named) == 0 {
			not = append(not, "no account here is declared never to become root, so agents run as the operator's "+
				"account, which may become root")
		} else if len(not) == 0 {
			confined = "the account " + strings.Join(named, ", ") + " cannot become root without a person (judged " +
				st.At.UTC().Format(time.RFC3339) + ")"
		}
	}
	if served {
		not = append(not, "the login shell runs any command an agent gives it as the machine's runtime account, which "+
			"can become root ("+servedWhy+")")
	}
	if len(not) > 0 {
		v.Why = strings.Join(not, "; ")
		return v
	}
	v.Free, v.Why = true, confined+"; the login shell's execute is not served here"
	return v
}

func orNoneKnownWord(s string) string {
	if s == "" {
		return "is not judged"
	}
	return s
}

// settingFileSuffix is the resource the login shell's holder renders its setting execute into (ADR 0268: zsh's
// tools-conf, `execute=serve` or `execute=withhold`), read beside the bus's discovery.
const settingFileSuffix = ".tools-conf"

// executeServedHere is whether the login shell's execute is served on this machine (ADR 0268), failing closed: served
// while the bus's discovery hears it answered here or could not be asked, or while the holder's setting file applied
// here says anything but withhold or cannot be read. why says which.
func executeServedHere(statePath string, heard func() (bool, error)) (bool, string) {
	var why []string
	if heard == nil {
		why = append(why, "the bus could not be asked whether execute is answered here")
	} else if h, err := heard(); err != nil {
		why = append(why, "the bus could not be asked whether execute is answered here: "+err.Error())
	} else if h {
		why = append(why, "the bus hears execute answered here")
	}
	known, err := store.Load(statePath)
	if err != nil {
		why = append(why, "what this machine applied could not be read: "+err.Error())
	}
	for _, r := range known.Resources {
		if r.Type != "file" || !strings.HasSuffix(r.ID, settingFileSuffix) {
			continue
		}
		if value, err := settingIn(r.Target); err != nil {
			why = append(why, moduleOf(r.ID)+"'s setting execute cannot be read: "+err.Error())
		} else if value != "withhold" {
			why = append(why, moduleOf(r.ID)+"'s setting execute here is "+value)
		}
	}
	return len(why) > 0, strings.Join(why, "; ")
}

// settingIn is the value of execute in a setting file, "" when it says none.
func settingIn(path string) (string, error) {
	f, err := os.Open(path)
	if err != nil {
		return "", err
	}
	defer f.Close()
	value := ""
	lines := bufio.NewScanner(f)
	for lines.Scan() {
		t := strings.TrimSpace(lines.Text())
		if key, v, ok := strings.Cut(t, "="); ok && !strings.HasPrefix(t, "#") && strings.TrimSpace(key) == "execute" {
			value = strings.TrimSpace(v)
		}
	}
	if value == "" {
		value = "not said"
	}
	return value, lines.Err()
}

// keepWaiting is every library a look found waiting for the operator's word to keep it (ADR 0328 §8).
func keepWaiting(vs []link.ResourceHealth) []string {
	var out []string
	for _, v := range vs {
		for _, w := range v.AccessWaits {
			if w.Reason == link.WaitKeep {
				out = append(out, w.Access)
			}
		}
	}
	return out
}

// engineActs is the node seat this engine holds (ADR 0563): judged and performed by the same code as the controller's
// signed asks, its asks to the operator over the session's link, its acts kept beside the state.
func engineActs(opts options, node string, queue *link.Session, search *accounts.SetuidCache) *link.HandActs {
	state := link.BusState{Session: queue, Node: node}
	h := &link.HandActs{Node: node, State: state,
		JudgeHandOver: handOverJudgeFrom(opts.state), HandOver: handOverFrom(opts.state),
		JudgeAccess: accessJudgeFrom(opts.state), AccessAct: accessActFrom(opts.state, opts.timeout),
		JudgeDirectories: judgeDirectoriesFrom(opts.state), SetuidSearch: queue.SetuidSearch, Searching: search.Running,
		RootFree: func() link.RootVerdict {
			served, why := executeServedHere(opts.state, func() (bool, error) {
				return queue.HeardServed("node-login-shell", "execute")
			})
			return judgeRootFree(node, lastAccounts.Load(), time.Now(), served, why)
		},
		ManifestReads: queue.StatedReads, Say: queue.SayEngine}
	asker := occ.EngineAsker(node)
	h.Asker = &occ.Client{Module: asker, State: state, Publish: queue.PublishAsk,
		Record: func(id string) (*stdio.StateEntry, error) { return queue.RouterRecord(asker + "." + id) }}
	return h
}

// reconcileEvery is how often the engine catches up its open asks from the router's records (a warrant heard while no
// link was open, an ask past its expiry): besides each link's opening.
const reconcileEvery = 5 * time.Minute

func reconcileActs(ctx context.Context, h *link.HandActs) {
	t := time.NewTicker(reconcileEvery)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			h.Reconcile()
		}
	}
}

// compositionReadFor bounds one read of the verb composition: the revisions of one key from the bucket.
const compositionReadFor = 20 * time.Second

// compositionFrom answers the verb composition from the node agent that runs now (issue 774); refused in words while
// none does.
func compositionFrom(current *agentNow) func(module, digest string, diff bool) (any, error) {
	return func(module, digest string, diff bool) (any, error) {
		ag := current.get()
		if ag == nil {
			return nil, errors.New("this engine's node agent has not started, so it holds no composition yet")
		}
		ctx, cancel := context.WithTimeout(context.Background(), compositionReadFor)
		defer cancel()
		return ag.Composition(ctx, module, digest, diff)
	}
}

// composeReadFor bounds the verb compose's wait for its pass: a pass that waits on a seat longer says what it waits on.
const composeReadFor = 2 * time.Minute

// composeFrom answers the verb compose (issue 774): a pass of the composer that runs now, and with module that module's
// apply tried again by the node agent; refused in words while the node composes nothing of its own.
func composeFrom(current *agentNow, composer *atomic.Pointer[selfcompose.Composer]) func(module string) (any, error) {
	return func(module string) (any, error) {
		c := composer.Load()
		if c == nil {
			return nil, errors.New("this node composes none of its own modules now (it applies the controller's " +
				"compositions, or its node agent is starting), so there is no pass to make")
		}
		out := map[string]any{}
		if module != "" {
			// Only a failed apply that is not held is tried again; what runs is never taken away (review round 1: the
			// verb has no warrant, ADR 0563 §2, and a hold for a failed backup is the operator's to release, ADR 0568 §4).
			ag := current.get()
			if ag == nil {
				return nil, errors.New("this engine's node agent has not started, so no apply is tried again")
			}
			if err := ag.RetryFailed(module); err != nil {
				return nil, err
			}
			out["retried"] = module
		}
		ctx, cancel := context.WithTimeout(context.Background(), composeReadFor)
		defer cancel()
		pass, err := c.ComposeNow(ctx)
		if err != nil {
			return nil, err
		}
		out["pass"] = pass
		return out, nil
	}
}

// packagesReadFor bounds the verb packages' reads and its write.
const packagesReadFor = 20 * time.Second

// packagesFrom answers the verb packages (novox/hq ADR 0565 §6, §15): this node's key of the engine's state packages;
// with write, the key written again from what the
// composer that runs now handed, refused in words while the node composes none of its own.
func packagesFrom(node string, session *link.Session, composer *atomic.Pointer[selfcompose.Composer]) func(write bool) (any, error) {
	return func(write bool) (any, error) {
		ctx, cancel := context.WithTimeout(context.Background(), packagesReadFor)
		defer cancel()
		var wrote *bool
		if write {
			c := composer.Load()
			if c == nil {
				return nil, errors.New("this node composes none of its own modules now (its node agent is starting), " +
					"so it has no packages of its own to write")
			}
			w, err := c.WritePackages(ctx)
			if err != nil {
				return nil, fmt.Errorf("this node's key of %s is not written: %w", link.PackagesBucket, err)
			}
			wrote = &w
		}
		key, keyErr := link.EnginePackages{Session: session}.Get(ctx, node)
		v := link.PackagesAnswer(node, key, keyErr)
		v.Wrote = wrote
		return v, nil
	}
}

// composeCandidateFrom answers the verb compose-candidate (novox/hq ADR 0593 §7): the check's candidate composed on this
// node in shadow by the run's shadow, applying and writing nothing, within the shadow's own bound (shorter than
// nec.AnswerBound, selfcompose candidateWithin). The answer is
// the seat's shape (nodeengine.ComposeCandidateAnswer), checked as it is encoded; a node whose shadow does not run yet
// answers unready, which the checker takes as no answer.
func composeCandidateFrom(node string, shadow *atomic.Pointer[selfcompose.Shadow]) func(candidate string, input bool) (any, error) {
	return func(candidate string, input bool) (any, error) {
		ask := nodeengine.ComposeCandidateInput{Candidate: candidate, Input: input}
		if _, err := nec.EncodeAsk(ask); err != nil {
			return nil, err
		}
		var a nodeengine.ComposeCandidateAnswer
		if s := shadow.Load(); s == nil {
			// Unready, never an error the checker reads words of: its agent is starting (ADR 0593 §9).
			a = nodeengine.ComposeCandidateAnswer{Node: node, Candidate: candidate, Verdict: nec.Unready,
				Refused: "this node's shadow does not run now (its node agent is starting), so it composes no candidate"}
		} else {
			ctx, cancel := context.WithTimeout(context.Background(), nec.AnswerBound)
			defer cancel()
			a = s.ComposeCandidate(ctx, ask)
		}
		raw, err := nec.EncodeAnswer(a)
		if err != nil {
			return nil, fmt.Errorf("this node's answer to %s does not fit the seat's shape: %w", candidate, err)
		}
		return json.RawMessage(raw), nil
	}
}