mesh-catalog/modules/disk-load/cmd/disk-load/keeper.go

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/health"
	"os"
	"os/exec"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
)

// Store is the state `readings` as the keeper uses it.
type Store interface {
	Get(key string) (json.RawMessage, error)
	Put(key string, value any) error
	Keys() ([]string, error)
}

// busStore is one of disk-load's states through the runtime: readings, or rounds.
type busStore struct{ state string }

func (b busStore) Get(key string) (json.RawMessage, error) {
	e, err := stdio.State(b.state).Get(key)
	if err != nil || e == nil {
		return nil, err
	}
	return e.Value, nil
}
func (b busStore) Put(key string, v any) error {
	_, err := stdio.State(b.state).Put(key, v)
	return err
}
func (b busStore) Keys() ([]string, error) { return stdio.State(b.state).Keys() }

// Keeper measures the items of its node and keeps the readings.
type Keeper struct {
	Node string
	// Items is every data item the modules placed on this node declare, as the node-engine composed them and its state
	// data holds them (Watched.All, novox/hq ADR 0577 clause 2).
	Items    func(ctx context.Context) (map[string][]catalogue.PlacedDataItem, error)
	Measurer Measurer
	Store    Store
	// Rounds is the state rounds, where each round says how far it got (ADR 0577 clause 3); nil keeps none.
	Rounds Store
	Say    func(string, ...any)

	mu sync.Mutex // one round at a time

	// pending are the modules whose items changed since they were last measured (Due), measured by Work.
	pmu     sync.Mutex
	pending map[string]bool
	wake    chan struct{}

	// unwatched says a round was refused because the watch had not caught up (ErrNotWatchedYet): one runs as soon as
	// it has (Watched.Follow), not an hour later.
	unwatched atomic.Bool
}

// RoundIfRefused runs one round now when the last was refused for a watch not caught up, and says whether it did.
func (k *Keeper) RoundIfRefused(ctx context.Context) bool {
	if !k.unwatched.CompareAndSwap(true, false) {
		return false
	}
	_, _, _ = k.Round(ctx, false)
	return true
}

// SayEvery is how many items a round measures between two lines saying how far it got.
const SayEvery = 50

// RoundVersion is the version of a round's layout.
const RoundVersion = 1

// Round is the value of a node's key of the state rounds: when its last round started and ended, how many items it
// had to measure, how many it measured, wrote and kept, and each item that failed and why. Ended is nil while it runs;
// Error says why a round measured nothing at all.
type Round struct {
	V        int          `json:"v"`
	Node     string       `json:"node"`
	Walks    bool         `json:"walks,omitempty"`
	Started  time.Time    `json:"started"`
	Ended    *time.Time   `json:"ended,omitempty"`
	Items    int          `json:"items"`
	Measured int          `json:"measured"`
	Written  int          `json:"written"`
	Kept     int          `json:"kept"`
	Failed   []FailedItem `json:"failed,omitempty"`
	Error    string       `json:"error,omitempty"`
}

// FailedItem is one item a round could not measure or keep, `<module>.<item>`, and why.
type FailedItem struct {
	Item string `json:"item"`
	Why  string `json:"why"`
}

// putRound writes the round to this node's key; a round that cannot be written is said, and measures on.
func (k *Keeper) putRound(r Round) {
	if k.Rounds == nil {
		return
	}
	if err := k.Rounds.Put(k.Node, r); err != nil {
		k.Say("the round's outcome cannot be kept in the state %s: %v", StateRounds, err)
	}
}

// Round measures every item that is not a cache, one at a time, and writes the readings that changed. An item
// measured by a walk is walked only when walks is true or its last walk is older than WalkEvery: the hourly round
// reads counters, and a walk runs at most once a day (ADR 0233 §22).
//
// A round says how far it got (novox/hq ADR 0577 clause 3): it writes its outcome to this node's key of the state rounds
// when it starts, every SayEvery items and when it ends, and says the same in the journal. An item that fails is a
// failed item of the round, never its end.
func (k *Keeper) Round(ctx context.Context, walks bool) (written, kept int, err error) {
	k.mu.Lock()
	defer k.mu.Unlock()
	r := Round{V: RoundVersion, Node: k.Node, Walks: walks, Started: k.Measurer.Now().UTC()}
	end := func() {
		at := k.Measurer.Now().UTC()
		r.Ended = &at
		k.putRound(r)
	}
	k.putRound(r)
	k.Say("a round of measuring the data declared here starts (walks: %v)", walks)
	items, err := k.Items(ctx)
	if err != nil {
		if errors.Is(err, ErrNotWatchedYet) {
			k.unwatched.Store(true)
		}
		r.Error = err.Error()
		end()
		k.Say("the round measured nothing: %v", err)
		return 0, 0, err
	}
	modules := make([]string, 0, len(items))
	for m := range items {
		modules = append(modules, m)
		for _, it := range items[m] {
			if it.Class != ClassCache {
				r.Items++
			}
		}
	}
	sort.Strings(modules)
	var failed []string
	for _, module := range modules {
		for _, it := range items[module] {
			if it.Class == ClassCache {
				continue
			}
			w, said, err := k.measureOne(ctx, it, walks)
			r.Measured++
			switch {
			case err != nil:
				failed = append(failed, fmt.Sprintf("%s.%s: %v", module, it.ID, err))
				r.Failed = append(r.Failed, FailedItem{Item: module + "." + it.ID, Why: err.Error()})
			case said != "":
				// Measured, and the measurement failed: its reading says so, and the round counts it failed.
				r.Failed = append(r.Failed, FailedItem{Item: module + "." + it.ID, Why: said})
			}
			if err == nil && w {
				written++
			} else if err == nil {
				kept++
			}
			r.Written, r.Kept = written, kept
			if r.Measured%SayEvery == 0 {
				k.Say("the round has measured %d of %d items: %d written, %d unchanged, %d failed", r.Measured, r.Items,
					written, kept, len(r.Failed))
				k.putRound(r)
			}
		}
	}
	// A round retires nothing: a reading is retired only when its module's key says so (MeasureModule), never because
	// an item is missing from what the watch holds now, which may be empty or partial.
	end()
	k.Say("the round ended: %d of %d items measured, %d readings written, %d unchanged, %d failed", r.Measured, r.Items,
		written, kept, len(r.Failed))
	if len(failed) > 0 {
		return written, kept, errors.New("readings not kept: " + strings.Join(failed, "; "))
	}
	return written, kept, nil
}

// Due asks Work to measure a module's items: its key of the engine's state data changed (novox/hq ADR 0577 clause 2).
func (k *Keeper) Due(module string) {
	k.pmu.Lock()
	if k.pending == nil {
		k.pending, k.wake = map[string]bool{}, make(chan struct{}, 1)
	}
	k.pending[module] = true
	wake := k.wake
	k.pmu.Unlock()
	select {
	case wake <- struct{}{}:
	default:
	}
}

// Work measures each module Due names, one at a time, until ctx ends.
func (k *Keeper) Work(ctx context.Context) {
	k.pmu.Lock()
	if k.pending == nil {
		k.pending, k.wake = map[string]bool{}, make(chan struct{}, 1)
	}
	wake := k.wake
	k.pmu.Unlock()
	for {
		select {
		case <-ctx.Done():
			return
		case <-wake:
		}
		for {
			k.pmu.Lock()
			var module string
			for m := range k.pending {
				module = m
				break
			}
			delete(k.pending, module)
			k.pmu.Unlock()
			if module == "" {
				break
			}
			k.MeasureModule(ctx, module)
		}
	}
}

// MeasureModule measures one module's items now, as its key of the engine's state data changed, and retires the
// readings of a module that no longer declares any here. A walk is not repeated within a day for an item that did not
// change (measureOne).
func (k *Keeper) MeasureModule(ctx context.Context, module string) {
	k.mu.Lock()
	defer k.mu.Unlock()
	items, err := k.Items(ctx)
	if err != nil {
		k.Say("%s's data items changed and are not measured: %v", module, err)
		return
	}
	n, err := k.retireGone(module, items[module])
	if err != nil {
		k.Say("%s's readings of items it no longer declares here are not retired: %v", module, err)
	} else if n > 0 {
		k.Say("%s no longer declares %d item(s) here: their readings are retired, measured no more", module, n)
	}
	if len(items[module]) == 0 {
		return
	}
	written, kept := 0, 0
	var failed []string
	for _, it := range items[module] {
		if it.Class == ClassCache {
			continue
		}
		w, said, err := k.measureOne(ctx, it, false)
		switch {
		case err != nil:
			failed = append(failed, it.ID+": "+err.Error())
		case said != "":
			failed = append(failed, it.ID+": "+said)
		case w:
			written++
		default:
			kept++
		}
	}
	k.Say("%s's data items changed and were measured: %d readings written, %d unchanged, %d failed%s", module, written,
		kept, len(failed), map[bool]string{true: " (" + strings.Join(failed, "; ") + ")", false: ""}[len(failed) > 0])
}

// retireGone marks retired every reading of this node of module whose item its key of the engine's state data no longer
// names — the key was deleted (the module was unassigned there, or is no longer pulled in) or names fewer items — with
// when and why, and keeps it: the record of what the module left behind, until a person deletes it (ADR 0233). It is
// asked only on a change of that module's key (novox/hq ADR 0577 clause 2), never from a round. It is cleared when the
// item is declared again (measureOne).
func (k *Keeper) retireGone(module string, declared []catalogue.PlacedDataItem) (int, error) {
	return k.retire(func(r health.Reading) bool { return r.Module == module }, declared)
}

// RetireUnkeyed retires, once the watch handed over every current key, the readings of this node whose module has no
// key in that set: a key deleted while the watch was not up (disk-load restarted, the machine rebooted or slept) never
// reaches disk-load as a delete. An empty set is skipped and said: the engine has not written its keys yet, and an
// empty set would retire every reading.
func (k *Keeper) RetireUnkeyed(ctx context.Context) {
	k.mu.Lock()
	defer k.mu.Unlock()
	items, err := k.Items(ctx)
	if err != nil {
		k.Say("the readings of modules with no key are not looked for: %v", err)
		return
	}
	if len(items) == 0 {
		k.Say("the node-engine's state %s holds no key of this machine yet: no reading is retired for it", delivery.EngineDataBucket)
		return
	}
	var declared []catalogue.PlacedDataItem
	for _, its := range items {
		declared = append(declared, its...)
	}
	n, err := k.retire(func(r health.Reading) bool { _, keyed := items[r.Module]; return !keyed }, declared)
	switch {
	case err != nil:
		k.Say("the readings of modules with no key are not all retired: %v", err)
	case n > 0:
		k.Say("%d reading(s) of modules with no key of this machine retired, measured no more", n)
	}
}

// retire marks retired every reading of this node that of picks and declared does not name.
func (k *Keeper) retire(of func(health.Reading) bool, declared []catalogue.PlacedDataItem) (int, error) {
	keys, err := k.Store.Keys()
	if err != nil {
		return 0, err
	}
	declaredNow := map[string]bool{}
	for _, it := range declared {
		declaredNow[health.ReadingKey(k.Node, it.Module, it.ID)] = true
	}
	now := k.Measurer.Now().UTC()
	n := 0
	for _, key := range keys {
		if !strings.HasPrefix(key, k.Node+".") || declaredNow[key] {
			continue
		}
		raw, err := k.Store.Get(key)
		if err != nil || len(raw) == 0 {
			continue
		}
		r, err := health.DecodeReading(raw)
		if err != nil || r.Node != k.Node || !of(r) || !r.RetiredAt.IsZero() {
			continue
		}
		r.RetiredAt = now
		r.RetiredWhy = fmt.Sprintf("no longer declared on %s: its module was unassigned there, or is no longer pulled in", k.Node)
		if err := k.Store.Put(key, r); err != nil {
			return n, err
		}
		n++
	}
	return n, nil
}

// measureOne measures one item, unless it is a walk done within WalkEvery of an unchanged item, and writes its reading
// when it changed. It answers whether it wrote the reading, what the measurement said failed ("" when it did not, or
// when the item was not measured now), and an error when the reading could not be read or kept.
func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, walks bool) (bool, string, error) {
	key := health.ReadingKey(k.Node, it.Module, it.ID)
	var prev *health.Reading
	raw, err := k.Store.Get(key)
	if err != nil {
		return false, "", err
	}
	if len(raw) > 0 {
		if p, err := health.DecodeReading(raw); err == nil {
			prev = &p
		}
	}
	now := k.Measurer.Now()
	if prev != nil && prev.RetiredAt.IsZero() && prev.PlacedDataItem == it && it.Measure == catalogue.MeasureWalk && !walks &&
		prev.MeasuredAt != nil && now.Sub(*prev.MeasuredAt) < WalkEvery {
		return false, "", nil
	}
	m := k.Measurer.Measure(ctx, it)
	next, changed := health.NextReading(prev, k.Node, it, m, now)
	if prev != nil && !prev.RetiredAt.IsZero() {
		// Declared again: it is no longer retired.
		next.RetiredAt, next.RetiredWhy, changed = time.Time{}, "", true
	}
	if !changed {
		return false, m.Error, nil
	}
	return true, m.Error, k.Store.Put(key, next)
}

// MeasureNow measures one item now, walk or not, and writes its reading when it changed.
func (k *Keeper) MeasureNow(ctx context.Context, module, item string) (health.Reading, bool, error) {
	k.mu.Lock()
	defer k.mu.Unlock()
	items, err := k.Items(ctx)
	if err != nil {
		return health.Reading{}, false, err
	}
	for _, it := range items[module] {
		if it.ID != item {
			continue
		}
		if it.Class == ClassCache {
			return health.Reading{}, false, fmt.Errorf("%s.%s is a cache, which is listed and never measured", module, item)
		}
		w, _, err := k.measureOne(ctx, it, true)
		if err != nil {
			return health.Reading{}, false, err
		}
		raw, err := k.Store.Get(health.ReadingKey(k.Node, module, item))
		if err != nil {
			return health.Reading{}, w, err
		}
		r, err := health.DecodeReading(raw)
		return r, w, err
	}
	return health.Reading{}, false, fmt.Errorf("%s declares no data item %q on this machine", module, item)
}

// Loop runs a round after a few minutes and then every `every`, with the walks of the night's round when it is the
// hour after the night; ctx ends it.
func (k *Keeper) Loop(ctx context.Context, every time.Duration, first time.Duration) {
	select {
	case <-ctx.Done():
		return
	case <-time.After(first):
	}
	for {
		_, _, _ = k.Round(ctx, false) // it says its outcome itself, in the journal and its key of the state rounds
		select {
		case <-ctx.Done():
			return
		case <-time.After(every):
		}
	}
}

// RootRunner runs a command as root, through sudo without a prompt when this process is not root. A walk is bounded
// by WalkFor itself; the runner's own bound is a little longer.
func RootRunner(ctx context.Context, name string, args ...string) (string, error) {
	ctx, cancel := context.WithTimeout(ctx, WalkFor+time.Minute)
	defer cancel()
	program, argv := name, args
	if os.Getuid() != 0 {
		program, argv = "sudo", append([]string{"-n", name}, args...)
	}
	var stdout, stderr bytes.Buffer
	cmd := exec.CommandContext(ctx, program, argv...)
	cmd.Stdout, cmd.Stderr = &stdout, &stderr
	if err := cmd.Run(); err != nil {
		said := strings.TrimSpace(stderr.String())
		if said == "" {
			said = strings.TrimSpace(stdout.String())
		}
		if program == "sudo" && strings.HasPrefix(said, "sudo:") {
			return "", fmt.Errorf("%s needs root and the runtime's account may not run it without a prompt: %s", name, said)
		}
		lines := strings.Split(said, "\n")
		if len(lines) > 3 {
			lines = lines[len(lines)-3:]
		}
		if said == "" {
			return "", fmt.Errorf("%s: %w", name, err)
		}
		return "", errors.New(strings.Join(lines, " / "))
	}
	return stdout.String(), nil
}