Merge branch 'feat/774-l3-reland-leaves' into merge-leaves

This commit is contained in:
2026-10-30 00:01:12 +01:00
68 changed files with 1656 additions and 941 deletions
+1
View File
@@ -0,0 +1 @@
disk-load
+248
View File
@@ -0,0 +1,248 @@
package main
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/health"
"os"
"os/exec"
"sort"
"strings"
"sync"
"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 the state through the runtime.
type busStore struct{}
func (busStore) Get(key string) (json.RawMessage, error) {
e, err := stdio.State(StateReadings).Get(key)
if err != nil || e == nil {
return nil, err
}
return e.Value, nil
}
func (busStore) Put(key string, v any) error {
_, err := stdio.State(StateReadings).Put(key, v)
return err
}
func (busStore) Keys() ([]string, error) { return stdio.State(StateReadings).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, composed from the bus (frombus.DataItems).
Items func(ctx context.Context) (map[string][]catalogue.PlacedDataItem, error)
Measurer Measurer
Store Store
Say func(string, ...any)
mu sync.Mutex // one round at a time
}
// 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).
func (k *Keeper) Round(ctx context.Context, walks bool) (written, kept int, err error) {
k.mu.Lock()
defer k.mu.Unlock()
items, err := k.Items(ctx)
if err != nil {
return 0, 0, err
}
modules := make([]string, 0, len(items))
for m := range items {
modules = append(modules, m)
}
sort.Strings(modules)
var failed []string
for _, module := range modules {
for _, it := range items[module] {
if it.Class == ClassCache {
continue
}
w, err := k.measureOne(ctx, it, walks)
if err != nil {
failed = append(failed, fmt.Sprintf("%s.%s: %v", module, it.ID, err))
continue
}
if w {
written++
} else {
kept++
}
}
}
retired, err := k.retireGone(items)
written += retired
if err != nil {
failed = append(failed, err.Error())
}
if len(failed) > 0 {
return written, kept, errors.New("readings not kept: " + strings.Join(failed, "; "))
}
return written, kept, nil
}
// retireGone marks retired every reading of this node whose item the node no longer declares — its module was unassigned
// there, or is no longer pulled in — with when and why, and keeps it: the record of what the module left behind, until a
// person deletes it (ADR 0233). It is cleared when the item is declared again (measureOne).
func (k *Keeper) retireGone(declared map[string][]catalogue.PlacedDataItem) (int, error) {
keys, err := k.Store.Keys()
if err != nil {
return 0, err
}
declaredNow := map[string]bool{}
for module, its := range declared {
for _, it := range its {
declaredNow[health.ReadingKey(k.Node, 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 || !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
}
func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, walks bool) (bool, 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() && 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, nil
}
return true, 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 {
w, kept, err := k.Round(ctx, false)
if err != nil {
k.Say("measuring the data declared here: %v", err)
} else {
k.Say("measured the data declared here: %d readings changed, %d unchanged", w, kept)
}
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
}
+43 -2
View File
@@ -1,20 +1,61 @@
// The disk-load module's Go bundle (novox/hq ADR 0188, ADR 0193, issue 312): a process the node's
// runtime launches and speaks MCP over stdio to. It reads the machine's storage load from /proc and
// /sys — iowait, pressure stall, each disk's busy share, throughput, IOPS, wait and queue depth, the
// filesystems' space and the processes doing the I/O — and changes nothing.
// filesystems' space and the processes doing the I/O — and changes nothing of the machine.
//
// It also measures the data items the modules of its node declare (ADR 0233) and keeps each reading and the item's
// seven-day peak as its own state `readings` (ADR 0571 clause 3), composed from the bus (no file).
package main
import (
"context"
"fmt"
"os"
"strings"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/announce"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/compose/fromruntime"
)
func logf(format string, args ...any) { fmt.Fprintf(os.Stderr, "[disk-load] "+format+"\n", args...) }
func main() {
if err := stdio.Keeps(announce.State{Name: StateReadings}); err != nil {
logf("%v", err)
os.Exit(1)
}
node := os.Getenv("MESH_NODE")
k := &Keeper{Node: node, Store: busStore{}, Say: logf, Measurer: Measurer{Run: RootRunner, Now: time.Now},
Items: func(ctx context.Context) (map[string][]catalogue.PlacedDataItem, error) {
return placedItems(ctx, node)
}}
if node != "" {
every := time.Hour
if d, err := time.ParseDuration(os.Getenv("MESH_DISK_MEASURE_EVERY")); err == nil && d >= time.Minute {
every = d
}
go k.Loop(context.Background(), every, 3*time.Minute)
}
tools := append(Tools(Here()), ReadingTools(k)...)
// An empty name serves as the module the runtime names (MESH_SERVED_MODULE): disk-load.
if err := stdio.Serve("", Tools(Here())); err != nil {
if err := stdio.Serve("", tools); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
}
// placedItems is every data item the modules placed on node declare, composed from the bus as the node composes itself.
func placedItems(ctx context.Context, node string) (map[string][]catalogue.PlacedDataItem, error) {
items, unplaced, err := frombus.DataItems(ctx, fromruntime.Reader(), frombus.Mesh{}, frombus.Scope{Node: node})
if err != nil {
return nil, err
}
if len(unplaced) > 0 {
logf("data items left out as not placeable: %s", strings.Join(unplaced, "; "))
}
return items, nil
}
+215
View File
@@ -0,0 +1,215 @@
package main
// How an item is measured (novox/hq ADR 0233 §22 to §25), moved here from the backup holder, which only reads the
// readings now (ADR 0571 clause 3). Never an unbounded walk of something large.
import (
"context"
"fmt"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/health"
"regexp"
"strconv"
"strings"
"time"
)
// The bounds on a walk: a walk is for small items, and one that meets a large one stops and says so.
var (
WalkFor = 10 * time.Minute
WalkFiles = 2_000_000
WalkEvery = 23 * time.Hour
)
// walkCommand sums the files under a path and finds its newest change, in one walk, bounded by time and by files.
// One line out: "<bytes> <epoch seconds> <files>", then the status of find.
func walkCommand(path string) string {
return fmt.Sprintf("timeout %d find '%s' -xdev -printf '%%y %%s %%T@\\n' 2>/dev/null | head -n %d | "+
"awk 'BEGIN{s=0;m=0;n=0} {n++; if ($1==\"f\") s+=$2; if ($3>m) m=$3} END {printf \"%%d %%.0f %%d\\n\", s, m, n}'; "+
"echo \"${PIPESTATUS[0]:-0}\"", int(WalkFor.Seconds()), path, WalkFiles)
}
// topCommand is the newest change among a path and its top-level entries, and how many there are.
func topCommand(path string) string {
return "timeout 60 find '" + path + "' -maxdepth 1 -printf '%T@\\n' 2>/dev/null | " +
"awk 'BEGIN{m=0;n=0} {n++; if ($1>m) m=$1} END {printf \"%.0f %d\\n\", m, n-1}'"
}
func epochTime(s string) *time.Time {
epoch, err := strconv.ParseInt(s, 10, 64)
if err != nil || epoch <= 0 {
return nil
}
t := time.Unix(epoch, 0).UTC()
return &t
}
// Measurer measures items through a runner that has the root the walks need.
type Measurer struct {
Run Runner
Now func() time.Time
}
func (b Measurer) exists(ctx context.Context, path string) bool {
_, err := b.Run(ctx, "test", "-e", path)
return err == nil
}
// Measure is one item, measured now, by its own method.
func (b Measurer) Measure(ctx context.Context, it catalogue.PlacedDataItem) health.Measured {
at := b.Now().UTC()
m := health.Measured{MeasuredAt: &at}
if !b.exists(ctx, it.Path) {
m.Error = it.Path + " does not exist"
zero := int64(0)
m.SizeBytes, m.Precision = &zero, "exact"
return m
}
m.Redundancy = b.redundancyOf(ctx, it.Path)
switch it.Measure {
case catalogue.MeasureDataset, catalogue.MeasureShallow:
out, err := b.Run(ctx, "bash", "-c", topCommand(it.Path))
if f := strings.Fields(out); err == nil && len(f) == 2 {
m.LastWrite = epochTime(f[0])
} else if err != nil {
m.Error = err.Error()
}
m.Precision = "no size: the newest change among its top-level entries only"
if it.Measure == catalogue.MeasureShallow {
return m
}
out, err = b.Run(ctx, "zfs", "list", "-Hp", "-o", "name,used", it.Path)
f := strings.Fields(out)
if err != nil || len(f) != 2 {
m.Precision = "no size: it is on no ZFS dataset to read one from; the newest change among its top-level entries only"
return m
}
used, err := strconv.ParseInt(f[1], 10, 64)
if err != nil {
return m
}
m.SizeBytes = &used
m.Precision = "dataset " + f[0] + ": its whole size, from the filesystem's counters; the newest change among the top-level entries"
return m
}
out, err := b.Run(ctx, "bash", "-c", walkCommand(it.Path))
if err != nil {
m.Error = err.Error()
return m
}
lines := strings.Split(strings.TrimSpace(out), "\n")
f := strings.Fields(lines[0])
if len(f) != 3 || len(lines) < 2 {
m.Error = "the measurement said " + strings.TrimSpace(out)
return m
}
size, err1 := strconv.ParseInt(f[0], 10, 64)
files, err2 := strconv.Atoi(f[2])
if err1 != nil || err2 != nil {
m.Error = "the measurement said " + strings.TrimSpace(out)
return m
}
m.SizeBytes, m.LastWrite = &size, epochTime(f[1])
m.Precision = "exact"
if status := strings.TrimSpace(lines[len(lines)-1]); status == "124" || files >= WalkFiles {
m.Precision = fmt.Sprintf("partial: measured partially — stopped after %s or %d files; the size is a lower bound, "+
"and this item wants measure: dataset or shallow", WalkFor, WalkFiles)
}
return m
}
// redundancyOf is the redundant storage a path is on: a ZFS pool's health, an md array's members, a btrfs
// filesystem's error counters. Nil for storage with no redundancy it knows how to read.
func (b Measurer) redundancyOf(ctx context.Context, path string) *health.Redundancy {
out, err := b.Run(ctx, "findmnt", "-no", "SOURCE,FSTYPE", "--target", path)
if err != nil {
return nil
}
f := strings.Fields(out)
if len(f) < 2 {
return nil
}
source, fstype := f[0], f[1]
yes, no := true, false
switch {
case fstype == "zfs":
pool, _, _ := strings.Cut(source, "/")
r := &health.Redundancy{Kind: "zfs", Where: pool}
health, err := b.Run(ctx, "zpool", "list", "-H", "-o", "health", pool)
if err != nil {
r.Said = "zpool list: " + err.Error()
return r
}
status, err := b.Run(ctx, "zpool", "status", "-x", pool)
if err != nil {
r.Said = "zpool status: " + err.Error()
return r
}
health, status = strings.TrimSpace(health), strings.TrimSpace(status)
r.Said = "health " + health + "; " + firstLineOf(status)
if health == "ONLINE" && strings.Contains(status, "is healthy") {
r.Healthy = &yes
} else {
r.Healthy = &no
r.Said = "health " + health + "; " + oneLineOf(status)
}
return r
case strings.HasPrefix(source, "/dev/md"):
device := strings.TrimPrefix(source, "/dev/")
r := &health.Redundancy{Kind: "md", Where: device}
mdstat, err := b.Run(ctx, "cat", "/proc/mdstat")
if err != nil {
r.Said = err.Error()
return r
}
healthy, said, found := mdHealth(mdstat, device)
if !found {
r.Said = device + " is not in /proc/mdstat"
return r
}
r.Said = said
if healthy {
r.Healthy = &yes
} else {
r.Healthy = &no
}
return r
case fstype == "btrfs":
r := &health.Redundancy{Kind: "btrfs", Where: source}
stats, err := b.Run(ctx, "btrfs", "device", "stats", "--check", path)
if err != nil {
r.Healthy, r.Said = &no, "device errors: "+oneLineOf(err.Error())
return r
}
r.Healthy, r.Said = &yes, firstLineOf(stats)
return r
}
return nil
}
var mdMembers = regexp.MustCompile(`\[([U_]+)\]`)
// mdHealth reads one array's block of /proc/mdstat: healthy when every member is up and it is not recovering.
func mdHealth(mdstat, device string) (bool, string, bool) {
lines := strings.Split(mdstat, "\n")
for i, l := range lines {
if !strings.HasPrefix(l, device+" ") {
continue
}
block := l
for j := i + 1; j < len(lines) && strings.HasPrefix(lines[j], " "); j++ {
block += " " + strings.TrimSpace(lines[j])
}
members := mdMembers.FindStringSubmatch(block)
healthy := members != nil && !strings.Contains(members[1], "_") && !strings.Contains(block, "recovery")
return healthy, oneLineOf(block), true
}
return false, "", false
}
func firstLineOf(s string) string {
line, _, _ := strings.Cut(strings.TrimSpace(s), "\n")
return line
}
func oneLineOf(s string) string { return strings.Join(strings.Fields(s), " ") }
@@ -0,0 +1,12 @@
package main
// The readings layout, their key and the peak are mesh-sdk's (health.Reading, health.NextReading; novox/hq ADR 0571
// clause 3): disk-load measures each placed data item of its node and keeps the readings as its own state.
import "git.novox.be/novox/mesh-sdk/go/health"
// StateReadings is the state disk-load keeps.
const StateReadings = health.ReadingsState
// ClassCache is the class of an item that is listed and never measured.
const ClassCache = "cache"
@@ -0,0 +1,144 @@
package main
import (
"context"
"encoding/json"
"os"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/health"
)
type memStore struct{ m map[string]json.RawMessage }
func (s *memStore) Get(k string) (json.RawMessage, error) { return s.m[k], nil }
func (s *memStore) Put(k string, v any) error {
raw, _ := json.Marshal(v)
s.m[k] = raw
return nil
}
func (s *memStore) Keys() ([]string, error) {
var out []string
for k := range s.m {
out = append(out, k)
}
return out, nil
}
// A round measures what the data file declares but caches, keeps the readings, writes nothing the second time, and a
// walk is not repeated within a day unless asked.
func TestARoundMeasuresTheDeclaredItemsAndWritesWhatChanged(t *testing.T) {
text := "# pg\nitem data valuable /srv/pg /srv/pg backup walk\nitem tmp cache /srv/tmp - none\n# bad-line-module\nitem logs rebuildable /srv/logs - none shallow\n"
placed, err := catalogue.ParseDataItems(text)
if err != nil {
t.Fatal(err)
}
walks := 0
run := func(_ context.Context, name string, args ...string) (string, error) {
switch {
case name == "test":
return "", nil
case name == "findmnt":
return "", os.ErrNotExist
case name == "bash" && strings.Contains(args[1], "-maxdepth 1"):
return "1760000000 2\n", nil
case name == "bash":
walks++
return "4096 1760000000 3\n0\n", nil
}
return "", os.ErrNotExist
}
now := time.Date(2026, 10, 29, 12, 0, 0, 0, time.UTC)
store := &memStore{m: map[string]json.RawMessage{}}
k := &Keeper{Node: "ace", Items: func(context.Context) (map[string][]catalogue.PlacedDataItem, error) { return placed, nil }, Store: store, Say: func(string, ...any) {},
Measurer: Measurer{Run: run, Now: func() time.Time { return now }}}
w, kept, err := k.Round(context.Background(), false)
if err != nil || w != 2 || kept != 0 || len(store.m) != 2 {
t.Fatalf("first round: %d written %d kept %v, keys %d", w, kept, err, len(store.m))
}
now = now.Add(time.Hour)
if w, kept, _ := k.Round(context.Background(), false); w != 0 || kept != 2 || walks != 1 {
t.Fatalf("second round: %d written, %d kept, %d walks; wanted none written and the walk not repeated", w, kept, walks)
}
if _, _, err := k.MeasureNow(context.Background(), "pg", "data"); err != nil || walks != 2 {
t.Fatalf("measure now: %v, %d walks", err, walks)
}
// An item no longer declared is kept, retired with when and why; declared again, it is not retired any more.
delete(placed, "pg")
if _, _, err := k.Round(context.Background(), false); err != nil {
t.Fatal(err)
}
gone, _ := health.DecodeReading(store.m["ace.pg.data"])
if gone.RetiredAt.IsZero() || !strings.Contains(gone.RetiredWhy, "no longer declared on ace") || gone.FirstSeen.IsZero() {
t.Fatalf("a reading of an item no longer declared: %+v", gone)
}
placed["pg"] = []catalogue.PlacedDataItem{{Module: "pg", ID: "data", Class: "valuable", Path: "/srv/pg", Measure: catalogue.MeasureWalk}}
if _, _, err := k.Round(context.Background(), false); err != nil {
t.Fatal(err)
}
if back, _ := health.DecodeReading(store.m["ace.pg.data"]); !back.RetiredAt.IsZero() {
t.Fatalf("a reading declared again is still retired: %+v", back)
}
if _, _, err := k.MeasureNow(context.Background(), "pg", "tmp"); err == nil {
t.Fatal("a cache was measured")
}
}
// The redundant storage under an item is read: a ZFS pool healthy, then degraded; an md array with a member missing; and
// plain storage, which is no redundancy at all. A dataset is measured from the filesystem's counters and a shallow item
// from its top-level entries only; a walk that meets more than its bound stops and says so.
func TestTheStorageUnderAnItemAndTheLargeMethodsAreRead(t *testing.T) {
poolHealth, statusX, fs := "ONLINE\n", "pool 'storage' is healthy\n", "storage/media zfs\n"
run := func(_ context.Context, name string, args ...string) (string, error) {
line := name + " " + strings.Join(args, " ")
switch {
case name == "test":
return "", nil
case name == "findmnt":
return fs, nil
case name == "zpool" && args[0] == "list":
return poolHealth, nil
case name == "zpool":
return statusX, nil
case name == "cat":
return "Personalities : [raid1]\nmd127 : active raid1 sdb1[1] sda1[0]\n 976630464 blocks super 1.2 [2/1] [U_]\n\nunused devices: <none>\n", nil
case name == "zfs" && args[len(args)-1] == "/storage/media/movies":
return "storage/media\t97067690722560\n", nil
case name == "zfs":
return "", os.ErrNotExist
case name == "bash" && strings.Contains(line, "-maxdepth 1"):
return "1759744800 12673\n", nil
case name == "bash":
return "123 1759700000 2000000\n141\n", nil
}
return "", nil
}
m := Measurer{Run: run, Now: time.Now}
ctx := context.Background()
got := m.Measure(ctx, catalogue.PlacedDataItem{Module: "media", ID: "movies", Path: "/storage/media/movies", Measure: catalogue.MeasureDataset})
if got.SizeBytes == nil || *got.SizeBytes != 97067690722560 || !strings.HasPrefix(got.Precision, "dataset storage/media") ||
got.LastWrite == nil || got.Redundancy == nil || got.Redundancy.Kind != "zfs" || got.Redundancy.Healthy == nil || !*got.Redundancy.Healthy {
t.Fatalf("the library from its dataset: %+v", got)
}
poolHealth, statusX = "DEGRADED\n", " pool: storage\n state: DEGRADED\nstatus: One or more devices has been removed\n"
if r := m.redundancyOf(ctx, "/storage/media/movies"); r.Healthy == nil || *r.Healthy || !strings.Contains(r.Said, "DEGRADED") {
t.Fatalf("a degraded pool: %+v", r)
}
fs = "/dev/md127 ext4\n"
if r := m.redundancyOf(ctx, "/srv/x"); r == nil || r.Kind != "md" || r.Healthy == nil || *r.Healthy {
t.Fatalf("an array missing a member: %+v", r)
}
fs = "/dev/nvme0n1p3 ext4\n"
if r := m.redundancyOf(ctx, "/var/lib/x"); r != nil {
t.Fatalf("plain storage read as redundant: %+v", r)
}
if s := m.Measure(ctx, catalogue.PlacedDataItem{Path: "/mnt/plex/config", Measure: catalogue.MeasureShallow}); s.SizeBytes != nil || !strings.HasPrefix(s.Precision, "no size") {
t.Fatalf("a shallow item: %+v", s)
}
if w := m.Measure(ctx, catalogue.PlacedDataItem{Path: "/srv/big", Measure: catalogue.MeasureWalk}); !strings.HasPrefix(w.Precision, "partial") {
t.Fatalf("a walk stopped at its bound: %+v", w)
}
}
@@ -0,0 +1,93 @@
package main
import (
"context"
"fmt"
"sort"
"strings"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/health"
)
// ReadingTools are the tools over the state `readings`: every key, one key, how full the state is, and one item
// measured now (a module's tools are its command line).
func ReadingTools(k *Keeper) []stdio.Tool {
return []stdio.Tool{
{
Name: "disk_readings",
Description: "The data readings disk-load keeps for this machine's data items (ADR 0233, ADR 0571 clause 3): every key " +
"`<node>.<module>.<item>` with its size, newest write, measured-at, precision and seven-day peak, or one key with " +
"all it holds when `key` is given; and how full the state is: keys kept, their bytes, and the cap the bus puts on " +
"a state (none declared). Filter by `module`. (r)",
Input: map[string]any{
"key": map[string]any{"type": "string", "description": "one key, <node>.<module>.<item>, with everything it holds"},
"module": map[string]any{"type": "string", "description": "only this module's items"},
},
Run: func(args map[string]any) (any, error) {
if key := text(args, "key"); key != "" {
raw, err := k.Store.Get(key)
if err != nil {
return nil, err
}
if len(raw) == 0 {
return nil, fmt.Errorf("no reading is kept at %q", key)
}
r, err := health.DecodeReading(raw)
return r, err
}
keys, err := k.Store.Keys()
if err != nil {
return nil, err
}
sort.Strings(keys)
module := text(args, "module")
type line struct {
Key string `json:"key"`
SizeBytes *int64 `json:"size_bytes,omitempty"`
PeakBytes *int64 `json:"peak_bytes,omitempty"`
Precision string `json:"precision,omitempty"`
Error string `json:"error,omitempty"`
Reading health.Reading `json:"-"`
}
var lines []line
total := 0
for _, key := range keys {
raw, err := k.Store.Get(key)
if err != nil {
return nil, err
}
total += len(raw)
r, err := health.DecodeReading(raw)
if err != nil || (module != "" && r.Module != module) {
continue
}
lines = append(lines, line{Key: key, SizeBytes: r.SizeBytes, PeakBytes: r.Peak, Precision: r.Precision, Error: r.Error})
}
return map[string]any{"readings": lines, "fill": map[string]any{"keys": len(keys), "bytes": total,
"cap": "none declared: the bus's own limit on a state applies"}}, nil
},
},
{
Name: "disk_measure",
Description: "Measure one data item of this machine now, whatever its method (a walk too), and write its reading when it " +
"changed. Reads the disks as root: a walk is bounded by time and files and says when it was partial.",
Input: map[string]any{
"module": map[string]any{"type": "string", "description": "the module that declares the item"},
"item": map[string]any{"type": "string", "description": "the item's id"},
},
Run: func(args map[string]any) (any, error) {
module, item := text(args, "module"), text(args, "item")
if module == "" || item == "" {
return nil, fmt.Errorf("module and item are required")
}
r, written, err := k.MeasureNow(context.Background(), module, item)
if err != nil {
return nil, err
}
return map[string]any{"written": written, "reading": r, "key": health.ReadingKey(k.Node, module, item),
"note": strings.TrimSpace(map[bool]string{true: "the reading changed and was written", false: "the reading had not changed"}[written])}, nil
},
},
}
}
@@ -3,6 +3,7 @@ package main
import (
"encoding/json"
"os"
"regexp"
"sort"
"strings"
"testing"
@@ -35,9 +36,11 @@ func TestTheManifestNamesExactlyTheToolsTheBundleServesAndNoMachine(t *testing.T
t.Fatal(err)
}
var served []string
for _, tl := range Tools(newFake(t).machine()) {
k := &Keeper{Store: &memStore{m: map[string]json.RawMessage{}}}
for _, tl := range append(Tools(newFake(t).machine()), ReadingTools(k)...) {
served = append(served, tl.Name)
if !strings.HasSuffix(tl.Description, "(r)") {
// disk_measure writes the module's own state of readings and nothing else of the machine.
if !strings.HasSuffix(tl.Description, "(r)") && tl.Name != "disk_measure" {
t.Errorf("%s only reads and does not say so", tl.Name)
}
}
@@ -50,7 +53,7 @@ func TestTheManifestNamesExactlyTheToolsTheBundleServesAndNoMachine(t *testing.T
t.Errorf("the module only reads, and declares %d resources", len(m.Resources))
}
for _, banned := range []string{"g14", "novox", "ace", "shanks", "jochen", "/home/"} {
if strings.Contains(string(raw), banned) {
if regexp.MustCompile(`(^|[^a-z0-9-])` + regexp.QuoteMeta(banned) + `($|[^a-z0-9-])`).Match(raw) {
t.Errorf("the manifest says %q", banned)
}
}
+2 -2
View File
@@ -1,5 +1,5 @@
module diskload
go 1.22
go 1.26.0
require git.novox.be/novox/mesh-sdk/go v0.1.6
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029225606-31e90aaa0f0d
+14 -2
View File
@@ -1,2 +1,14 @@
git.novox.be/novox/mesh-sdk/go v0.1.6 h1:9qzdYONYbJdWcu6sxQcq9v1LI0JxcfkiKYkMUzJSkVQ=
git.novox.be/novox/mesh-sdk/go v0.1.6/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029225606-31e90aaa0f0d h1:HHSXxRS+GEYNgtdg9MFB8W2Y6ycPJF7YLF46y3oj7Ow=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029225606-31e90aaa0f0d/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
+15 -2
View File
@@ -7,7 +7,9 @@
"disk_filesystems",
"disk_folders",
"disk_devices",
"cpu_top"
"cpu_top",
"disk_readings",
"disk_measure"
],
"build": {
"artifacts": [
@@ -23,5 +25,16 @@
]
}
]
}
},
"state": [
"readings"
],
"reads": [
"mesh-config.resolved",
"mesh-delivery-assigner.placement",
"mesh-delivery-builder.builds",
"mesh-delivery-indexer.manifests",
"mesh-delivery-indexer.modules",
"node-engine.have"
]
}
@@ -9,7 +9,7 @@ package main
//
// - every image a container on this machine uses, in any state;
// - every image this machine's declaration names — what the mesh would send it now and what it was
// last sent — as the controller's `images` verb answers. **No answer, no removal**: what the
// last sent — as the mesh-artifact-sweep seat's `images` verb answers. **No answer, no removal**: what the
// declaration names is the one thing that cannot be guessed;
// - every image younger than the floor (seven days, the weekly prune's week);
// - in each line of images holding one of those, the newest image besides them: the previous
@@ -43,7 +43,7 @@ type PruneImagesAsk struct {
Limit int
}
// declaredImages is the controller's `images` answer (novox/hq ADR 0251 §5).
// declaredImages is the mesh-artifact-sweep seat's `images` answer (novox/hq ADR 0251 §5).
type declaredImages struct {
Node string `json:"node"`
Images []struct {
@@ -129,7 +129,7 @@ func digestOf(ref string) string {
return ""
}
// askDeclared asks the controller what this machine's declarations name. An error means unknown.
// askDeclared asks the mesh-artifact-sweep seat what this machine's declarations name. An error means unknown.
func (c *Client) askDeclared() (*declaredImages, error) {
if c.Node == "" {
return nil, errors.New("the runtime did not say which machine this is (MESH_NODE is empty)")
@@ -137,20 +137,20 @@ func (c *Client) askDeclared() (*declaredImages, error) {
if c.Ask == nil {
return nil, errors.New("this bundle cannot ask the mesh")
}
raw, err := c.Ask("seat:mesh-controller.images", map[string]any{"node": c.Node})
raw, err := c.Ask("seat:mesh-artifact-sweep.images", map[string]any{"node": c.Node})
if err != nil {
return nil, fmt.Errorf("the controller's images verb did not answer: %w", err)
return nil, fmt.Errorf("the mesh-artifact-sweep seat's images verb did not answer: %w", err)
}
answer, output, ok := answerOf(raw)
if !ok {
return nil, fmt.Errorf("the controller refused images: %s", firstLine(output))
return nil, fmt.Errorf("the mesh-artifact-sweep seat refused images: %s", firstLine(output))
}
var d declaredImages
if err := json.Unmarshal(answer, &d); err != nil {
return nil, fmt.Errorf("the controller's images answer is not readable: %v", err)
return nil, fmt.Errorf("the mesh-artifact-sweep seat's images answer is not readable: %v", err)
}
if d.Node != "" && d.Node != c.Node {
return nil, fmt.Errorf("the controller answered for %q, not for this machine %q", d.Node, c.Node)
return nil, fmt.Errorf("the mesh-artifact-sweep seat answered for %q, not for this machine %q", d.Node, c.Node)
}
return &d, nil
}
@@ -43,7 +43,7 @@ func imagesMachine() *fake {
func declaring(t *testing.T, answer string) Asker {
return func(key string, body any) (json.RawMessage, error) {
if key != "seat:mesh-controller.images" {
if key != "seat:mesh-artifact-sweep.images" {
t.Errorf("asked %s", key)
}
if b, _ := body.(map[string]any); b["node"] != "laptop" {
+2 -2
View File
@@ -1,5 +1,5 @@
module docker
go 1.22
go 1.26.0
require git.novox.be/novox/mesh-sdk/go v0.1.6
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029225351-7bd0f6c2bdf2
+2 -2
View File
@@ -1,2 +1,2 @@
git.novox.be/novox/mesh-sdk/go v0.1.6 h1:9qzdYONYbJdWcu6sxQcq9v1LI0JxcfkiKYkMUzJSkVQ=
git.novox.be/novox/mesh-sdk/go v0.1.6/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029225351-7bd0f6c2bdf2 h1:LXFwlVh0fGcOTU41PNkZDknHF070F/mhVy8Te6wKr3Q=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029225351-7bd0f6c2bdf2/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+1 -1
View File
@@ -7,7 +7,7 @@
"privileged"
],
"invokes": [
"seat:mesh-controller.images"
"seat:mesh-artifact-sweep.images"
],
"claims": [
{
+1
View File
@@ -0,0 +1 @@
mesh-build-image
Binary file not shown.
+1
View File
@@ -0,0 +1 @@
mesh-build-node
Binary file not shown.
+66 -7
View File
@@ -45,13 +45,72 @@ export interface Built {
branches?: string[];
}
/** The events a build outcome reaches this module on: the build seat's, as it happens, and the control plane's replay. */
export const BUILD_EVENTS = [
// The build seat node-build-agent and the retired mesh-build-machine went with build-agent (novox/hq ADR 0524 §5,
// ADR 0525 §10): no build seat says `built` since R4g, so what the mesh holds is asked when this module starts.
// What the mesh already held, when this module asks what it missed (novox/hq ADR 0134).
"mesh-controller.built-before",
] as const;
/**
* Where a build outcome comes from now: the builder's log (novox/hq ADR 0571 clauses 1 and 9). The module asks the
* builder's tool for the builds that ended since a time, and again whenever the builder's state of done keys changes.
*/
export const BUILDER = "mesh-delivery-builder";
export const BUILDS_SINCE_TOOL = "builder_builds_since";
export const BUILDS_STATE = "mesh-delivery-builder.builds";
export const MANIFESTS_STATE = "mesh-delivery-indexer.manifests";
/** One entry of the builder's log, an ended build (mesh-sdk delivery.BuildLogEntry). */
export interface LogEntry {
event: string;
module: string;
fp: string;
repository: string;
directory: string;
commit: string;
ref?: string;
node?: string;
at: string;
/** done, failed or queued. */
outcome?: string;
/** The digest of the manifest built, read from the indexer's state of manifests. */
manifest?: string;
against?: string[];
made?: Made[];
}
/** The answer of the builder's tool: every ended build since a time, oldest first, and whether the log held more. */
export interface BuildsSince {
since: string;
builds: LogEntry[];
more?: boolean;
}
/**
* What a tool answers, whether bare or wrapped as the runtime wraps a tool's reply ({output} or {content:[{text}]}).
*/
export function answerOf<T>(raw: unknown): T {
const r = raw as { output?: unknown; content?: { text?: string }[] } | null;
if (r && typeof r.output === "string") return JSON.parse(r.output) as T;
if (r && Array.isArray(r.content) && typeof r.content[0]?.text === "string") return JSON.parse(r.content[0].text) as T;
if (typeof raw === "string") return JSON.parse(raw) as T;
return raw as T;
}
/**
* A log entry as the build outcome the catalogue places: a build that is done is placed with its manifest, what it
* made and what it was built against; one that failed is passed over (placeable says so); one still queued has not
* ended and is not taken in. `replay` is set by the caller for history.
*/
export function builtFrom(e: LogEntry, manifest: unknown, replay: boolean): Built | undefined {
if (e.outcome === "queued") return undefined;
return {
module: e.module,
commit: e.commit,
repository: e.repository,
path: e.directory,
ref: e.ref ?? "",
manifest,
made: e.made ?? [],
against: e.against ?? [],
replay,
...(e.outcome === "failed" ? { failed: "the builder's log says the build failed" } : {}),
};
}
/**
* Whether a build outcome is a module-version to place in the graph, and why not when it is not.
+77 -15
View File
@@ -9,9 +9,15 @@
// and appear the next time round, once whatever they were waiting for is registered. A chain and a
// diamond need no special handling, and nothing holds a plan.
import { on, emit } from "@novox/mesh-sdk/events";
import { emit } from "@novox/mesh-sdk/events";
import { broker } from "@novox/mesh-sdk/messaging";
import { invokeTool } from "@novox/mesh-sdk/tools";
import { state } from "@novox/mesh-sdk/state";
import { Graph } from "./store.js";
import { BUILD_EVENTS, placeable, type Built } from "./heard.js";
import {
answerOf, builtFrom, placeable, BUILDER, BUILDS_SINCE_TOOL, BUILDS_STATE, MANIFESTS_STATE,
type Built, type BuildsSince, type LogEntry,
} from "./heard.js";
const graph = Graph.fromEnv();
@@ -77,18 +83,74 @@ const placeTheBuild = async (event: { body: unknown }): Promise<void> => {
}
};
// As it happens — from the build seat that builds now, and the retired one while it still answers — and
// what the mesh already held when this module asked what it missed (novox/hq issue 712).
for (const heard of BUILD_EVENTS) await on(heard, placeTheBuild);
// **From the builder's log** (novox/hq ADR 0571 clauses 1 and 9). The builder keeps every build's end in a log of its
// own and answers the builds that ended since a time by its tool. This module asks at its start for what it missed (all
// the log holds, taken in as history: registered, announced to nobody), and again whenever the builder's state of done
// keys changes, for what ended since the last build it took in. Registering a build already held changes nothing, and a
// build is taken in once per run of this module.
const taken = new Set<string>();
let since = new Date(0).toISOString();
let caughtUp = false;
let asking: Promise<void> | undefined;
let askAgain = false;
// **And ask for what was built before this catalogue existed** (novox/hq 04-ISSUES/050).
//
// The queue above is durable, so nothing is missed once this is running. What it cannot have is
// what was announced before it first ran — and on a fresh mesh that is never arbitrary: the shared
// base, the store this runs on, and this module itself are each necessarily built BEFORE a
// catalogue exists to hear about them. The graph's foundation is the part it never sees.
//
// Asked on every start, not only the first. A catalogue cannot tell whether it has a gap, and the
// answer is idempotent: registering a build already held changes nothing and announces nothing.
// Asked AFTER subscribing, so a build arriving during the replay is not lost between the two.
const manifestOf = async (digest: string | undefined): Promise<unknown> => {
if (!digest) return {};
const held = await state<{ manifest?: string }>(MANIFESTS_STATE).get(digest);
if (!held?.value.manifest) return {};
return JSON.parse(held.value.manifest);
};
const takeIn = async (e: LogEntry, replay: boolean): Promise<void> => {
const key = `${e.module}/${e.fp}`;
if (taken.has(key)) return;
const built = builtFrom(e, await manifestOf(e.manifest), replay);
if (!built) return;
await placeTheBuild({ body: built });
taken.add(key);
};
const catchUp = async (): Promise<void> => {
for (;;) {
const answer = answerOf<BuildsSince>(
await invokeTool(broker(), BUILDER, BUILDS_SINCE_TOOL, { since }),
);
for (const e of answer.builds ?? []) {
await takeIn(e, !caughtUp);
if (e.at > since) since = e.at;
}
if (!answer.more) break;
}
caughtUp = true;
};
/** One ask at a time; a change heard while asking is asked for again once, so none is missed. */
const ask = (): Promise<void> => {
if (asking) {
askAgain = true;
return asking;
}
asking = (async () => {
try {
do {
askAgain = false;
await catchUp();
} while (askAgain);
} catch (err) {
console.error("mesh-catalog: the builder's builds since", since, "could not be taken in:", err);
} finally {
asking = undefined;
}
})();
return asking;
};
// Asked AFTER watching begins, so a build ending while the first ask runs asks again rather than being missed: the watch
// hands every current key first (the start's ask), then every change.
await state(BUILDS_STATE).watch(async (change) => {
if (!change.current) await ask();
});
await ask();
// **And say that this catalogue has caught up** (novox/hq 04-ISSUES/050).
await emit("catching-up", {});
+9 -4
View File
@@ -25,9 +25,7 @@
"secrets": {
"postgres-database": "${dir:state}/database.secret"
},
"consumes": [
"mesh-controller.built-before"
],
"consumes": [],
"emits": [
"registered",
"upgraded",
@@ -92,5 +90,12 @@
"from": "files"
}
]
}
},
"invokes": [
"mesh-delivery-builder.builder_builds_since"
],
"reads": [
"mesh-delivery-builder.builds",
"mesh-delivery-indexer.manifests"
]
}
+19 -8
View File
@@ -2,7 +2,7 @@ import { test } from "node:test";
import assert from "node:assert/strict";
import { readFile } from "node:fs/promises";
import { BUILD_EVENTS, placeable } from "../heard.js";
import { answerOf, builtFrom, placeable, BUILDER, BUILDS_SINCE_TOOL, BUILDS_STATE, MANIFESTS_STATE } from "../heard.js";
// A walk's build as the build seat says it (mesh-controller's link.BuildResult), here mailu's of novox/hq issue 712.
const walkBuild = {
@@ -20,14 +20,25 @@ const walkBuild = {
made: [{ name: "code", kind: "bundle", reference: "artifact-store://mailu/code/blobs/sha256:d041" }],
};
test("the catalogue hears no retired build seat, and what the mesh held (issue 712, issue 1887 E2)", async () => {
for (const retired of ["node-build-agent.built", "mesh-build-machine.built"]) {
assert.ok(!(BUILD_EVENTS as readonly string[]).includes(retired), `${retired} is still heard`);
}
assert.ok(BUILD_EVENTS.includes("mesh-controller.built-before"));
// Subscribing is not enough: the mesh grants and routes only what the manifest consumes.
test("the catalogue takes builds from the builder's log, and hears no controller event (ADR 0571 clause 9)", async () => {
const manifest = JSON.parse(await readFile(new URL("../module.json", import.meta.url), "utf8"));
for (const heard of BUILD_EVENTS) assert.ok(manifest.consumes.includes(heard), `${heard} is not consumed`);
assert.deepEqual(manifest.consumes, []);
assert.ok(manifest.invokes.includes(`${BUILDER}.${BUILDS_SINCE_TOOL}`), "the builder's tool is not invoked");
for (const read of [BUILDS_STATE, MANIFESTS_STATE]) assert.ok(manifest.reads.includes(read), `${read} is not read`);
// A log entry is placed as a build: done with its manifest, made and against; failed passed over; queued not taken in.
const entry = { event: "build.ended", module: "mailu", fp: "fp1", repository: walkBuild.repository, directory: "modules/mailu",
commit: walkBuild.commit, ref: "main", at: "2026-10-29T20:00:00Z", outcome: "done", manifest: "ab", against: ["img@sha256:1"],
made: walkBuild.made };
const done = builtFrom(entry, walkBuild.manifest, false);
assert.deepEqual(placeable(done!), { placed: "yes", module: "mailu", commit: walkBuild.commit });
assert.deepEqual([done!.path, done!.against, done!.replay], ["modules/mailu", ["img@sha256:1"], false]);
assert.equal(placeable(builtFrom({ ...entry, outcome: "failed" }, {}, true)!).placed, "no");
assert.equal(builtFrom({ ...entry, outcome: "queued" }, {}, true), undefined);
// The tool's answer is read bare or as the runtime wraps it.
const bare = { since: "x", builds: [entry] };
for (const raw of [bare, { output: JSON.stringify(bare) }, { content: [{ text: JSON.stringify(bare) }] }]) {
assert.equal(answerOf<typeof bare>(raw).builds[0].module, "mailu");
}
});
test("a walk's build is placed, under its module and commit", () => {
+30 -43
View File
@@ -15,6 +15,7 @@ import (
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
)
@@ -150,60 +151,46 @@ func roundTrip(node string) error {
}
}
// controlNodes reads which machines hold the controller and the bus.
// controlNodes reads which machines hold the controller and the bus, from the placement seat's state of who holds
// which seat (novox/hq ADR 0571 clause 4).
func controlNodes() ([]string, error) {
raw, err := stdio.Ask("seat:mesh-controller.seats", map[string]any{})
st := stdio.State(delivery.SeatHoldersRead)
names, err := st.Keys()
if err != nil {
return nil, err
}
return holdersOf(raw, "mesh-controller", "mesh-broker")
keys := map[string][]byte{}
for _, k := range names {
e, err := st.Get(k)
if err != nil {
return nil, err
}
if e != nil {
keys[k] = e.Value
}
}
return holderNodes(keys, "mesh-controller", "mesh-broker")
}
// holdersOf reads the seats verb's answer, bare or inside a tool reply, for the named seats' machines.
func holdersOf(raw json.RawMessage, seats ...string) ([]string, error) {
var answer struct {
Seats []struct {
Seat string `json:"seat"`
Holders []struct {
Node string `json:"node"`
} `json:"holders"`
} `json:"seats"`
Output string `json:"output"`
Content []struct {
Text string `json:"text"`
} `json:"content"`
}
if err := json.Unmarshal(raw, &answer); err != nil {
var s string
if json.Unmarshal(raw, &s) == nil {
return holdersOf(json.RawMessage(s), seats...)
}
return nil, fmt.Errorf("the seats answer is not JSON")
}
if len(answer.Seats) == 0 {
switch {
case answer.Output != "":
return holdersOf(json.RawMessage(answer.Output), seats...)
case len(answer.Content) > 0:
return holdersOf(json.RawMessage(answer.Content[0].Text), seats...)
}
return nil, fmt.Errorf("the seats answer lists no seat")
}
// holderNodes is the machines holding any of the seats, in the order the seats are named, each once.
func holderNodes(keys map[string][]byte, seats ...string) ([]string, error) {
var out []string
seen := map[string]bool{}
for _, s := range answer.Seats {
for _, want := range seats {
if s.Seat != want {
continue
}
for _, h := range s.Holders {
if !seen[h.Node] {
seen[h.Node] = true
out = append(out, h.Node)
}
for _, seat := range seats {
holders, err := delivery.HoldersOf(keys, seat, "")
if err != nil {
return nil, err
}
for _, h := range holders {
if !seen[h.Node] {
seen[h.Node] = true
out = append(out, h.Node)
}
}
}
if len(out) == 0 {
return nil, fmt.Errorf("the placement seat's state names no holder of %s", strings.Join(seats, " or "))
}
return out, nil
}
@@ -37,7 +37,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) {
if !reflect.DeepEqual(m.Consumes, []string{ControllerSeat + "." + HeartbeatEvent, "mesh-self-check.checked"}) {
t.Fatalf("consumes %v", m.Consumes)
}
if !reflect.DeepEqual(m.Invokes, []string{m.Module + ".watcher_ping", "seat:mesh-controller.seats"}) {
if !reflect.DeepEqual(m.Invokes, []string{m.Module + ".watcher_ping"}) {
t.Fatalf("invokes %v", m.Invokes)
}
if len(m.Claims) != 0 {
@@ -1,7 +1,6 @@
package main
import (
"encoding/json"
"errors"
"strings"
"testing"
@@ -214,16 +213,18 @@ func TestOnTheControlNodeItSaysItIsMisplaced(t *testing.T) {
}
}
func TestTheSeatsAnswerIsReadInEitherShape(t *testing.T) {
bare := `{"seats":[{"seat":"mesh-controller","holders":[{"node":"n1","module":"mesh-controller"}]},
{"seat":"mesh-broker","holders":[{"node":"n1"},{"node":"n2"}]},{"seat":"git","holders":[{"node":"n3"}]}]}`
wrapped, _ := json.Marshal(map[string]any{"content": []map[string]string{{"type": "text", "text": bare}}})
output, _ := json.Marshal(map[string]any{"ok": true, "output": bare})
for _, raw := range []string{bare, string(wrapped), string(output)} {
got, err := holdersOf(json.RawMessage(raw), "mesh-controller", "mesh-broker")
if err != nil || strings.Join(got, ",") != "n1,n2" {
t.Errorf("%s: %v %v", raw, got, err)
}
func TestTheHoldersAreReadFromThePlacementSeatsState(t *testing.T) {
keys := map[string][]byte{
"mesh-controller": []byte(`{"v":1,"seat":"mesh-controller","scope":"mesh","holders":[{"module":"mesh-controller","node":"n1"}]}`),
"mesh-broker": []byte(`{"v":1,"seat":"mesh-broker","scope":"mesh","holders":[{"module":"nats","node":"n1"},{"module":"nats","node":"n2"}]}`),
"git": []byte(`{"v":1,"seat":"git","scope":"mesh","holders":[{"module":"gitea","node":"n3"}]}`),
}
got, err := holderNodes(keys, "mesh-controller", "mesh-broker")
if err != nil || strings.Join(got, ",") != "n1,n2" {
t.Errorf("%v %v", got, err)
}
if _, err := holderNodes(map[string][]byte{}, "mesh-controller", "mesh-broker"); err == nil {
t.Error("no holder named, and no error")
}
}
+1 -1
View File
@@ -2,4 +2,4 @@ module mesh-watcher
go 1.26.0
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121
+2 -2
View File
@@ -1,2 +1,2 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a h1:elC2IjB+qbeXG+pWItLz71DenegbrdsqbcH5kyGo56Q=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121 h1:OsA1v6n736a7RqEV8Nkip8pf5ILMbzj0PWuKqcBOhK0=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+6 -4
View File
@@ -7,11 +7,11 @@
"mesh-self-check.checked"
],
"reads": [
"mesh-conditions.takeover"
"mesh-conditions.takeover",
"mesh-delivery-assigner.seat-holders"
],
"invokes": [
"mesh-watcher.watcher_ping",
"seat:mesh-controller.seats"
"mesh-watcher.watcher_ping"
],
"own-secrets": {
"telegram-token": "${dir:state}/telegram-token"
@@ -43,7 +43,9 @@
"type": "file",
"path": "${dir:state}/settings.json",
"mode": "0600",
"takes-config": ["telegram-chat-id"],
"takes-config": [
"telegram-chat-id"
],
"merge": "json",
"template": "files/settings.json"
}
@@ -20,7 +20,7 @@ func manifest(t *testing.T) map[string]any {
return m
}
// 1516/AC6: minio claims object-store (scope mesh) serving link, invokes seat:mesh-controller.seats, builds the Go
// 1516/AC6: minio claims object-store (scope mesh) serving link, reads the placement seat's seat-holders, builds the Go
// bundle, and no tool named minio_presigned_url remains.
func TestManifestClaimsTheSeat(t *testing.T) {
m := manifest(t)
@@ -38,9 +38,9 @@ func TestManifestClaimsTheSeat(t *testing.T) {
if c["name"] != "object-store" || c["scope"] != "mesh" || len(serves) != 1 || serves[0] != "link" {
t.Fatalf("claim: %v", c)
}
inv := m["invokes"].([]any)
if len(inv) != 1 || inv[0] != "seat:mesh-controller.seats" {
t.Fatalf("invokes: %v", inv)
reads, _ := m["reads"].([]any)
if _, invokes := m["invokes"]; invokes || len(reads) != 1 || reads[0] != "mesh-delivery-assigner.seat-holders" {
t.Fatalf("reads %v and invokes %v: it reads the placement seat's holders and calls nothing", reads, m["invokes"])
}
var goBundle map[string]any
for _, a := range m["build"].(map[string]any)["artifacts"].([]any) {
+33 -49
View File
@@ -13,6 +13,7 @@ import (
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/objectlink"
)
@@ -59,11 +60,11 @@ func linkKeyID(bucket string) string {
type server struct {
configPath, linksDir, receives string
now func() time.Time
// seats asks the controller for its seats; a test replaces it.
seats func() (json.RawMessage, error)
// seats reads the placement seat's state of who holds which seat, every key; a test replaces it.
seats func() (map[string][]byte, error)
mu sync.Mutex
holders map[string]bool // "<seat>/<node>.<module>"
keys map[string][]byte // seat-holders, as read
readAt time.Time
haveSet bool
}
@@ -77,8 +78,23 @@ func newServer(config, links, receives string) (*server, error) {
return &server{configPath: config, linksDir: links, receives: receives, now: time.Now, seats: askSeats}, nil
}
func askSeats() (json.RawMessage, error) {
return stdio.Ask("seat:mesh-controller.seats", map[string]any{})
func askSeats() (map[string][]byte, error) {
state := stdio.State(delivery.SeatHoldersRead)
names, err := state.Keys()
if err != nil {
return nil, err
}
keys := map[string][]byte{}
for _, k := range names {
e, err := state.Get(k)
if err != nil {
return nil, err
}
if e != nil {
keys[k] = e.Value
}
}
return keys, nil
}
// link answers one ask.
@@ -206,7 +222,7 @@ func (s *server) allowed(g grant, node, module, method string) error {
if seat, isSeat := strings.CutPrefix(sh.To, "seat:"); isSeat {
held, err := s.holds(seat, node, module)
if err != nil {
return false, fmt.Errorf("a share names seat %s and the controller's seats cannot be read to tell whether this caller holds it: %w", seat, err)
return false, fmt.Errorf("a share names seat %s and the placement seat's holders cannot be read to tell whether this caller holds it: %w", seat, err)
}
return held, nil
}
@@ -249,58 +265,26 @@ func has(list []string, v string) bool {
return false
}
// holds says whether node.module holds the seat, from the controller's seats, read at most once a minute. A read that
// failed is not cached: the next ask reads again.
// holds says whether node.module holds the seat, from the placement seat's state of who holds which seat (novox/hq ADR
// 0571 clause 4), read at most once a minute. A read that failed is not cached: the next ask reads again.
func (s *server) holds(seat, node, module string) (bool, error) {
s.mu.Lock()
defer s.mu.Unlock()
if !s.haveSet || s.now().Sub(s.readAt) >= holdersFor {
raw, err := s.seats()
keys, err := s.seats()
if err != nil {
return false, err
}
set, err := holdersIn(raw)
if err != nil {
return false, err
}
s.holders, s.readAt, s.haveSet = set, s.now(), true
s.keys, s.readAt, s.haveSet = keys, s.now(), true
}
return s.holders[seat+"/"+node+"."+module], nil
}
// holdersIn reads the controller's `seats` answer (answer.seats[].holders[] of {node, module}, and answer.outside[] of
// {Claim, Node, Module}) into "<seat>/<node>.<module>". The answer may arrive wrapped in {"answer": ...}.
func holdersIn(raw json.RawMessage) (map[string]bool, error) {
var doc struct {
Answer json.RawMessage `json:"answer"`
Seats []struct {
Seat string `json:"seat"`
Holders []struct {
Node string `json:"node"`
Module string `json:"module"`
} `json:"holders"`
} `json:"seats"`
Outside []struct {
Claim, Node, Module string
} `json:"outside"`
holders, err := delivery.HoldersOf(s.keys, seat, node)
if err != nil {
return false, err
}
if err := json.Unmarshal(raw, &doc); err != nil {
return nil, fmt.Errorf("the controller's seats answer is not JSON: %w", err)
}
if len(doc.Answer) > 0 && string(doc.Answer) != "null" {
return holdersIn(doc.Answer)
}
if doc.Seats == nil && doc.Outside == nil {
return nil, errors.New("the controller's seats answer holds neither seats nor outside claims")
}
out := map[string]bool{}
for _, s := range doc.Seats {
for _, h := range s.Holders {
out[s.Seat+"/"+h.Node+"."+h.Module] = true
for _, h := range holders {
if h.Node == node && h.Module == module {
return true, nil
}
}
for _, o := range doc.Outside {
out[o.Claim+"/"+o.Node+"."+o.Module] = true
}
return out, nil
return false, nil
}
@@ -1,7 +1,6 @@
package main
import (
"encoding/json"
"errors"
"net/url"
"os"
@@ -25,12 +24,13 @@ const grantFile = `{
]
}`
// The seats answer is the controller's (mesh-controller.seats read on 2026-10-21): seats[].holders[] of {node, module},
// and outside[] of {Claim, Node, Module}.
const seatsAnswer = `{"answer": {"seats": [
{"seat": "node-files", "scope": "node", "holders": [{"node": "g14", "module": "mesh-files"}, {"node": "shanks", "module": "mesh-files"}]},
{"seat": "node-backup", "scope": "node", "holders": [{"node": "g14", "module": "restic"}]}
], "outside": [{"Claim": "printer", "Scope": "mesh", "Node": "shanks", "Module": "brother", "Site": "home", "Kind": ""}]}}`
// The placement seat's state of who holds which seat (seat-holders): a key per node seat and node, a key per mesh seat.
var seatHolderKeys = map[string][]byte{
"node-files.g14": []byte(`{"v":1,"seat":"node-files","scope":"node","node":"g14","holders":[{"module":"mesh-files","node":"g14"}]}`),
"node-files.shanks": []byte(`{"v":1,"seat":"node-files","scope":"node","node":"shanks","holders":[{"module":"mesh-files","node":"shanks"}]}`),
"node-backup.g14": []byte(`{"v":1,"seat":"node-backup","scope":"node","node":"g14","holders":[{"module":"restic","node":"g14"}]}`),
"printer": []byte(`{"v":1,"seat":"printer","scope":"mesh","holders":[{"module":"brother","node":"shanks"}]}`),
}
type rig struct {
s *server
@@ -64,12 +64,12 @@ func newRig(t *testing.T) *rig {
t.Fatal(err)
}
s.now = func() time.Time { return r.now }
s.seats = func() (json.RawMessage, error) {
s.seats = func() (map[string][]byte, error) {
r.seatsCalls++
if r.seatsErr != nil {
return nil, r.seatsErr
}
return json.RawMessage(seatsAnswer), nil
return seatHolderKeys, nil
}
r.s = s
return r
+1 -1
View File
@@ -2,4 +2,4 @@ module minio
go 1.26.0
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021122348-6c857e1c025f
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121
+14 -2
View File
@@ -1,2 +1,14 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021122348-6c857e1c025f h1:7J/9jOfSfjfYpf8MOj96HwQRViQUkBNMb71RHP2mjn8=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021122348-6c857e1c025f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121 h1:OsA1v6n736a7RqEV8Nkip8pf5ILMbzj0PWuKqcBOhK0=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
+2 -2
View File
@@ -39,8 +39,8 @@
]
}
],
"invokes": [
"seat:mesh-controller.seats"
"reads": [
"mesh-delivery-assigner.seat-holders"
],
"contributes": {
"route": {
+8 -7
View File
@@ -14,6 +14,7 @@ package main
import (
"encoding/json"
"fmt"
"git.novox.be/novox/mesh-sdk/go/health"
"os"
"path/filepath"
"time"
@@ -122,22 +123,22 @@ func (m *Machine) leave(p string, v any) error {
// Asker calls a tool on the mesh as this module.
type Asker func(key string, body any) (json.RawMessage, error)
// DataVerb is the controller's verb that says where the mesh keeps data on a machine.
const DataVerb = "seat:mesh-controller.data"
// Readings reads disk-load's readings of one machine's data items (its state `readings`, ADR 0571 clause 3).
type Readings func(node string) ([]health.Reading, error)
// LeaveData asks the controller where the mesh keeps data on this machine and leaves the paths for the
// LeaveData reads disk-load for where the mesh keeps data on this machine and leaves the paths for the
// process (finding 4d). A failed ask keeps the last paths, with why; with none, the process sets up
// nothing new.
func (m *Machine) LeaveData(node string, ask Asker, now time.Time) []string {
if node == "" || ask == nil {
func (m *Machine) LeaveData(node string, read Readings, now time.Time) []string {
if node == "" || read == nil {
return nil
}
previous, _ := readJSON[DataPaths](m, DataFile)
out := DataPaths{Machine: node, Asked: now.Format(time.RFC3339)}
raw, err := ask(DataVerb, map[string]any{"machine": node})
readings, err := read(node)
var paths []string
if err == nil {
paths, err = DataFrom(raw)
paths, err = PathsFrom(readings)
}
var said []string
if err != nil {
+9 -37
View File
@@ -6,9 +6,9 @@ package main
// mesh keeps data in on this machine, from the controller's `data` verb over the bus.
import (
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-sdk/go/health"
"io"
"os"
"path/filepath"
@@ -63,44 +63,16 @@ func sortWants(w []Want) {
}
}
// DataFrom reads the paths out of mesh-controller.data's answer: `answer.data`, `data`, or the same as
// text in `output`. An answer with no data list is an error, never an empty list: an empty list would
// let every mount point through.
func DataFrom(raw json.RawMessage) ([]string, error) {
type item struct {
Path string `json:"path"`
}
var whole struct {
Answer *struct {
Data *[]item `json:"data"`
} `json:"answer"`
Data *[]item `json:"data"`
Output string `json:"output"`
}
if err := json.Unmarshal(raw, &whole); err != nil {
return nil, fmt.Errorf("the controller's data answer is not JSON: %v", err)
}
var items *[]item
switch {
case whole.Answer != nil && whole.Answer.Data != nil:
items = whole.Answer.Data
case whole.Data != nil:
items = whole.Data
case whole.Output != "":
var inner struct {
Data *[]item `json:"data"`
}
if err := json.Unmarshal([]byte(whole.Output), &inner); err == nil && inner.Data != nil {
items = inner.Data
}
}
if items == nil {
return nil, errors.New("the controller's data answer names no data list")
// PathsFrom is the paths disk-load's readings name: where the mesh keeps data here. No reading is an error, never an
// empty list: an empty list would let every mount point through.
func PathsFrom(readings []health.Reading) ([]string, error) {
if len(readings) == 0 {
return nil, errors.New("disk-load keeps no reading of this machine's data yet")
}
paths := []string{}
for _, it := range *items {
if strings.HasPrefix(it.Path, "/") {
paths = append(paths, filepath.Clean(it.Path))
for _, r := range readings {
if strings.HasPrefix(r.Path, "/") {
paths = append(paths, filepath.Clean(r.Path))
}
}
return paths, nil
+27 -2
View File
@@ -13,8 +13,10 @@ import (
"context"
"encoding/json"
"fmt"
"git.novox.be/novox/mesh-sdk/go/health"
"os"
"os/signal"
"strings"
"syscall"
"time"
@@ -62,13 +64,36 @@ func shareOnTheBus(m *Machine) {
_, err := stdio.State(SharesState).Put(key, value)
return err
}
ask := func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) }
read := func(node string) ([]health.Reading, error) {
st := stdio.State(health.ReadingsRead)
keys, err := st.Keys()
if err != nil {
return nil, err
}
var out []health.Reading
for _, k := range keys {
if !strings.HasPrefix(k, node+".") {
continue
}
e, err := st.Get(k)
if err != nil {
return nil, err
}
if e == nil {
continue
}
if r, err := health.DecodeReading(e.Value); err == nil {
out = append(out, r)
}
}
return out, nil
}
last, lastSaid := "", ""
for n := 0; ; n++ {
said := m.Share(node, get, put, &last, time.Now())
// Where the mesh keeps data here changes only with an assignment: asked at start and every five minutes.
if n%10 == 0 {
said = append(said, m.LeaveData(node, ask, time.Now())...)
said = append(said, m.LeaveData(node, read, time.Now())...)
}
if s := fmt.Sprint(said); len(said) > 0 && s != lastSaid {
fmt.Fprintln(os.Stderr, "mounts:", s)
+4 -6
View File
@@ -77,12 +77,10 @@ func TestTheManifestNamesWhatTheBundleServesAndWhatItNeeds(t *testing.T) {
if strings.Join(m.Requires, ",") != "nfs-share" {
t.Errorf("requires %v", m.Requires)
}
if strings.Join(m.Reads, ",") != ExportsState {
t.Errorf("the module reads nfs-server's exports from the bus: reads %v", m.Reads)
}
// It calls one verb: the controller's, for where the mesh keeps data on this machine (review finding 4d).
if strings.Join(m.Invokes, ",") != DataVerb {
t.Errorf("invokes %v, want %s", m.Invokes, DataVerb)
// It reads nfs-server's exports, and disk-load's readings for where the mesh keeps data on this machine (review
// finding 4d, ADR 0571 clause 3); it calls no verb.
if strings.Join(m.Reads, ",") != "disk-load.readings,"+ExportsState || len(m.Invokes) != 0 {
t.Errorf("reads %v, invokes %v", m.Reads, m.Invokes)
}
if !strings.Contains(raw, `node=${machine:name}`) {
t.Error("the config gives the process this machine's name, so it never mounts a share onto the machine that serves it")
+10 -19
View File
@@ -5,6 +5,8 @@ package main
import (
"context"
"encoding/json"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/health"
"os"
"path/filepath"
"strings"
@@ -198,18 +200,13 @@ func TestWithoutTheMeshsDataNothingNewIsSetUp(t *testing.T) {
func TestTheBundleLeavesTheMeshsDataForTheProcess(t *testing.T) {
f := newSystem(t)
f.write(ConfigFile, libraryOnly)
answer := json.RawMessage(`{"answer":{"data":[{"machine":"workstation","module":"plex","item":"data","path":"/mnt/plex/data"}]},"ok":true}`)
paths, err := DataFrom(answer)
rs := []health.Reading{{V: 1, Node: "workstation", PlacedDataItem: catalogue.PlacedDataItem{Module: "plex", ID: "data", Path: "/mnt/plex/data"}}}
paths, err := PathsFrom(rs)
if err != nil || len(paths) != 1 || paths[0] != "/mnt/plex/data" {
t.Fatalf("paths %q %v", paths, err)
}
// The controller's verbs also answer as text in output.
paths, err = DataFrom(json.RawMessage(`{"ok":true,"output":"{\"data\":[{\"path\":\"/srv/x\"}]}"}`))
if err != nil || len(paths) != 1 || paths[0] != "/srv/x" {
t.Fatalf("paths from output %q %v", paths, err)
}
if _, err := DataFrom(json.RawMessage(`{"ok":false}`)); err == nil {
t.Error("an answer with no data list is an error, not an empty list")
if _, err := PathsFrom(nil); err == nil {
t.Error("no reading is an error, not an empty list")
}
}
@@ -287,23 +284,17 @@ func callVerb(t *testing.T, m *Machine, name string, args map[string]any) (any,
func TestTheBundleKeepsTheLastDataWhenAnAskFails(t *testing.T) {
f := newSystem(t)
m := f.machine()
var asked []string
ok := func(key string, body any) (json.RawMessage, error) {
raw, _ := json.Marshal(body)
asked = append(asked, key+" "+string(raw))
return json.RawMessage(`{"answer":{"data":[{"path":"/mnt/plex/data"}]}}`), nil
ok := func(node string) ([]health.Reading, error) {
return []health.Reading{{V: 1, Node: node, PlacedDataItem: catalogue.PlacedDataItem{Module: "plex", ID: "data", Path: "/mnt/plex/data"}}}, nil
}
if said := m.LeaveData("workstation", ok, time.Now()); len(said) != 0 {
t.Fatalf("said %q", said)
}
if len(asked) != 1 || asked[0] != `seat:mesh-controller.data {"machine":"workstation"}` {
t.Errorf("asked %q", asked)
}
failing := func(string, any) (json.RawMessage, error) { return nil, os.ErrDeadlineExceeded }
failing := func(string) ([]health.Reading, error) { return nil, os.ErrDeadlineExceeded }
said := m.LeaveData("workstation", failing, time.Now())
d, err := readJSON[DataPaths](m, DataFile)
if err != nil || len(d.Paths) != 1 || d.Paths[0] != "/mnt/plex/data" || d.Error == "" || len(said) == 0 {
t.Errorf("a failed ask keeps the last paths and says why: %+v %v %q", d, err, said)
t.Errorf("a failed read keeps the last paths and says why: %+v %v %q", d, err, said)
}
}
+2 -2
View File
@@ -1,5 +1,5 @@
module mounts
go 1.22
go 1.26.0
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261010191001-33917f91ac3c
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029221656-196c5b6af277
+2 -8
View File
@@ -1,8 +1,2 @@
git.novox.be/novox/mesh-sdk/go v0.1.6 h1:9qzdYONYbJdWcu6sxQcq9v1LI0JxcfkiKYkMUzJSkVQ=
git.novox.be/novox/mesh-sdk/go v0.1.6/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
git.novox.be/novox/mesh-sdk/go v0.1.7 h1:C0sTQmtTiyYH7bnqZb7PusXnqA37gKuT7Nqjn9gG47w=
git.novox.be/novox/mesh-sdk/go v0.1.7/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261010190644-622b6b4f556d h1:BR8uwv4BqyLaBwITuLylrd9DRUnu0D5DjJw0RuQSXbk=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261010190644-622b6b4f556d/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261010191001-33917f91ac3c h1:l5onJwoIeH8yE/PjVlEeoPyXBsHp2NQiWLllz2fwX7s=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261010191001-33917f91ac3c/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029221656-196c5b6af277 h1:ajjsH4MWcx45MGYASjUSRQKhLzEringuk3TA0oxZRCU=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029221656-196c5b6af277/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+14 -16
View File
@@ -5,11 +5,9 @@
"nfs-share"
],
"reads": [
"disk-load.readings",
"nfs-server.exports"
],
"invokes": [
"seat:mesh-controller.data"
],
"capabilities": [
"package-manager",
"service-manager"
@@ -38,20 +36,20 @@
"per-machine": true
}
],
"config": {
"shares": {
"type": "string",
"description": "The shares of another machine this machine mounts: space-separated name=mountpoint, each optionally :ro, or none (novox/hq ADR 0263 rule 5).",
"trusted": true,
"default": "none"
},
"sources": {
"type": "string",
"description": "The occasional sources this machine mounts: space-separated name=smb://[<user>@]<host>/<share>@<mountpoint> or name=disk:<uuid>@<mountpoint>, each optionally :ro, or none (novox/hq ADR 0263 rule 6); an SMB source's username is in its entry, its password the secret smb-password-<name> (novox/hq ADR 0283).",
"trusted": true,
"default": "none"
}
"config": {
"shares": {
"type": "string",
"description": "The shares of another machine this machine mounts: space-separated name=mountpoint, each optionally :ro, or none (novox/hq ADR 0263 rule 5).",
"trusted": true,
"default": "none"
},
"sources": {
"type": "string",
"description": "The occasional sources this machine mounts: space-separated name=smb://[<user>@]<host>/<share>@<mountpoint> or name=disk:<uuid>@<mountpoint>, each optionally :ro, or none (novox/hq ADR 0263 rule 6); an SMB source's username is in its entry, its password the secret smb-password-<name> (novox/hq ADR 0283).",
"trusted": true,
"default": "none"
}
},
"own-secrets": {
"smb-password-*": {
"path": "${dir:state}/smb-password-*.secret",
+1 -1
View File
@@ -87,7 +87,7 @@ func tools(monitor Monitor, users func(context.Context) ([]byte, error), written
return Subscriptions(subsz, connz, arg(a, "subject"), limitOf(a, 100))
}},
{Name: "nats_streams",
Description: "JetStream: every stream with its subjects, retention, message count and size — or, given one stream, every consumer on it with what it has pending, unacknowledged and redelivered.",
Description: "JetStream: every stream with its subjects, retention, message count, size and size cap (max_bytes, when it has one) — or, given one stream, every consumer on it with what it has pending, unacknowledged and redelivered.",
Input: map[string]any{"stream": str("one stream, e.g. EVENTS, CONTROL, ASSIGNMENTS, SEAT_NODE_BUILD_AGENT or KV_<bucket>")},
Run: func(a map[string]any) (any, error) {
doc, err := get(jsz)
+4
View File
@@ -351,6 +351,10 @@ func Streams(jsz []byte, stream string) (map[string]any, error) {
if s.Config.MaxMsgsPerSubject > 0 {
row["max_per_subject"] = s.Config.MaxMsgsPerSubject
}
// The cap on the stream's size, so how full it is reads against it (-1 and 0 are no cap).
if s.Config.MaxBytes > 0 {
row["max_bytes"] = s.Config.MaxBytes
}
if stream != "" {
row["description"] = s.Config.Description
var consumers []map[string]any
+2 -2
View File
@@ -22,7 +22,7 @@ const subsz = `{"num_subscriptions":3,"subscriptions_list":[
{"subject":"mesh.node.anchor.declare","msgs":1,"cid":1}]}`
const jsz = `{"streams":2,"consumers":2,"messages":12,"bytes":300,"account_details":[{"stream_detail":[
{"name":"EVENTS","config":{"subjects":["mesh.mod.*.event.>"],"retention":"limits","max_age":604800000000000,"max_msgs_per_subject":10000},
{"name":"EVENTS","config":{"subjects":["mesh.mod.*.event.>"],"retention":"limits","max_age":604800000000000,"max_msgs_per_subject":10000,"max_bytes":1048576},
"state":{"messages":10,"bytes":200,"num_subjects":3,"consumer_count":2,"last_ts":"2026-10-04T10:00:00Z"},
"consumer_detail":[
{"name":"anchor_audit","config":{"filter_subject":"mesh.mod.*.event.>"},"num_pending":5,"num_ack_pending":1,"num_redelivered":2,"delivered":{"last_active":"2026-10-04T10:00:00Z"}},
@@ -80,7 +80,7 @@ func TestStreamsBacklogAndBuckets(t *testing.T) {
}
one, _ := Streams([]byte(jsz), "EVENTS")
row := one["detail"].([]map[string]any)[0]
if row["max_age_hours"].(int64) != 168 || len(row["consumer_detail"].([]map[string]any)) != 2 {
if row["max_age_hours"].(int64) != 168 || row["max_bytes"].(int64) != 1048576 || len(row["consumer_detail"].([]map[string]any)) != 2 {
t.Fatalf("%v", row)
}
if _, err := Streams([]byte(jsz), "NOPE"); err == nil {
+1
View File
@@ -0,0 +1 @@
nfs-server
+26 -48
View File
@@ -14,12 +14,12 @@ package main
// been read, what is exported stays exported and the status says the list is not known: the bus being
// slow is no reason to take the library away, and the look below still guards against a repository.
//
// The parsing of mesh-controller.data's answer is a small copy of the mounts module's (DataFrom). Shared
// code belongs in mesh-sdk: moving both there is mesh-sdk issue 13.
import (
"encoding/json"
"fmt"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
"os"
"path"
"sort"
@@ -46,68 +46,46 @@ type BackupFromBus struct {
Error string `json:"error,omitempty"`
}
// BackupFrom reads the holder of node-backup on one machine from mesh-controller.seats' answer, and the
// paths that module keeps data in on that machine from mesh-controller.data's. A machine whose backup seat
// nobody holds, and an answer with no data list, are errors: never an empty list read as "nothing to fear".
func BackupFrom(machine string, seats, data json.RawMessage) (string, []string, error) {
var s struct {
Answer *struct {
Seats []struct {
Seat string `json:"seat"`
Holders []struct {
Module string `json:"module"`
Node string `json:"node"`
} `json:"holders"`
} `json:"seats"`
} `json:"answer"`
}
if err := json.Unmarshal(seats, &s); err != nil || s.Answer == nil {
return "", nil, fmt.Errorf("the controller's seats answer could not be read: %v", err)
// BackupFrom reads the holder of node-backup on one machine from the placement seat's state of who holds which seat
// (novox/hq ADR 0571 clause 4), and the paths that module keeps data in on that machine from disk-load's readings
// (clause 3). A machine whose backup seat nobody holds, and no readings at all, are errors: never an empty list read as
// "nothing to fear".
func BackupFrom(machine string, keys map[string][]byte, readings []health.Reading) (string, []string, error) {
holders, err := delivery.HoldersOf(keys, BackupSeat, machine)
if err != nil {
return "", nil, err
}
holder := ""
for _, seat := range s.Answer.Seats {
if seat.Seat != BackupSeat {
continue
}
for _, h := range seat.Holders {
if h.Node == machine {
holder = h.Module
}
for _, h := range holders {
if h.Node == machine {
holder = h.Module
}
}
if holder == "" {
return "", nil, fmt.Errorf("no module holds %s on %s", BackupSeat, machine)
}
var d struct {
Answer *struct {
Data *[]struct {
Machine string `json:"machine"`
Module string `json:"module"`
Path string `json:"path"`
} `json:"data"`
} `json:"answer"`
}
if err := json.Unmarshal(data, &d); err != nil || d.Answer == nil || d.Answer.Data == nil {
return "", nil, fmt.Errorf("the controller's data answer holds no data list (%v)", err)
if len(readings) == 0 {
return "", nil, fmt.Errorf("disk-load keeps no reading of this machine's data yet")
}
var paths []string
for _, item := range *d.Answer.Data {
if item.Module == holder && (item.Machine == "" || item.Machine == machine) && path.IsAbs(item.Path) {
paths = append(paths, path.Clean(item.Path))
for _, r := range readings {
if r.Module == holder && (r.Node == "" || r.Node == machine) && path.IsAbs(r.Path) {
paths = append(paths, path.Clean(r.Path))
}
}
sort.Strings(paths)
return holder, paths, nil
}
// TakeBackup asks the controller over the bus, as the module, through the runtime.
// TakeBackup reads the bus, as the module, through the runtime.
func (m *Machine) TakeBackup(machine string) {
m.TakeBackupWith(machine, askBus)
m.TakeBackupWith(machine, readingsOf, seatHolderKeys)
}
// TakeBackupWith is TakeBackup with the ask given, for tests. A failed ask keeps the last holder and
// TakeBackupWith is TakeBackup with the ask and the reading of the seat-holders state given, for tests. A failed ask keeps the last holder and
// paths and adds why.
func (m *Machine) TakeBackupWith(machine string, ask func(key string, body any) (json.RawMessage, error)) {
func (m *Machine) TakeBackupWith(machine string, readings func(node string) ([]health.Reading, error),
holderKeys func() (map[string][]byte, error)) {
var last BackupFromBus
if text, err := m.Read(BackupFile); err == nil {
_ = json.Unmarshal([]byte(text), &last)
@@ -115,10 +93,10 @@ func (m *Machine) TakeBackupWith(machine string, ask func(key string, body any)
next := last
next.Looked = m.Now().UTC().Format(time.RFC3339)
next.Error = ""
seats, err := ask("seat:mesh-controller.seats", map[string]any{})
var data json.RawMessage
seats, err := holderKeys()
var data []health.Reading
if err == nil {
data, err = ask("seat:mesh-controller.data", map[string]any{"machine": machine})
data, err = readings(machine)
}
if err == nil {
var holder string
@@ -4,6 +4,8 @@ import (
"context"
"encoding/json"
"errors"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/health"
"strings"
"testing"
"time"
@@ -81,8 +83,8 @@ func TestWithoutTheBackupListTheExportsStayAndItIsSaid(t *testing.T) {
}
func TestBackupFromTheSeatsAndTheData(t *testing.T) {
seats := json.RawMessage(`{"answer":{"seats":[{"seat":"node-backup","scope":"node","holders":[{"module":"restic","node":"ace"},{"module":"borg","node":"shanks"}]}]},"ok":true}`)
data := json.RawMessage(`{"answer":{"data":[{"machine":"ace","module":"restic","item":"repository","path":"/storage/restic"},{"machine":"ace","module":"restic","item":"state","path":"/var/lib/restic"},{"machine":"ace","module":"plex","item":"data","path":"/mnt/plex/data"}]},"ok":true}`)
seats := backupHolders("ace", "restic")
data := readingsOfPaths("ace", map[string]string{"restic/repository": "/storage/restic", "restic/state": "/var/lib/restic", "plex/data": "/mnt/plex/data"})
holder, paths, err := BackupFrom("ace", seats, data)
if err != nil || holder != "restic" || strings.Join(paths, " ") != "/storage/restic /var/lib/restic" {
t.Errorf("got %q %v %v", holder, paths, err)
@@ -90,8 +92,8 @@ func TestBackupFromTheSeatsAndTheData(t *testing.T) {
if _, _, err := BackupFrom("g14", seats, data); err == nil {
t.Errorf("a machine whose backup seat nobody holds is not said")
}
if _, _, err := BackupFrom("ace", seats, json.RawMessage(`{"ok":true}`)); err == nil {
t.Errorf("an answer with no data list reads as an empty list")
if _, _, err := BackupFrom("ace", seats, nil); err == nil {
t.Errorf("no readings read as an empty list")
}
}
@@ -99,13 +101,13 @@ func TestBackupFromTheSeatsAndTheData(t *testing.T) {
func TestAFailedAskKeepsTheLastBackupPaths(t *testing.T) {
f := newFake(t)
m := f.machine()
m.TakeBackupWith("ace", func(string, any) (json.RawMessage, error) {
m.TakeBackupWith("ace", func(string) ([]health.Reading, error) {
return nil, errors.New("no responders")
})
}, noHolders)
f.backup("restic", "/storage/restic")
m.TakeBackupWith("ace", func(string, any) (json.RawMessage, error) {
m.TakeBackupWith("ace", func(string) ([]health.Reading, error) {
return nil, errors.New("timeout")
})
}, func() (map[string][]byte, error) { return backupHolders("ace", "restic"), nil })
var b BackupFromBus
if err := json.Unmarshal([]byte(f.read(BackupFile)), &b); err != nil {
t.Fatal(err)
@@ -152,15 +154,13 @@ func TestAStaleOrUnknownBackupListIsSaidWithItsAge(t *testing.T) {
t.Errorf("an unknown backup list is not a warning in the health: %+v", h)
}
ok := func(key string, body any) (json.RawMessage, error) {
if strings.Contains(key, "seats") {
return json.RawMessage(`{"answer":{"seats":[{"seat":"node-backup","holders":[{"module":"restic","node":"hs"}]}]}}`), nil
}
return json.RawMessage(`{"answer":{"data":[{"machine":"hs","module":"restic","path":"/srv/restic"}]}}`), nil
ok := func(string) ([]health.Reading, error) {
return readingsOfPaths("hs", map[string]string{"restic/repository": "/srv/restic"}), nil
}
m.TakeBackupWith("hs", ok)
m.TakeBackupWith("hs", ok, func() (map[string][]byte, error) { return backupHolders("hs", "restic"), nil })
now = now.Add(2 * time.Hour)
m.TakeBackupWith("hs", func(string, any) (json.RawMessage, error) { return nil, errors.New("the bus did not answer") })
m.TakeBackupWith("hs", func(string) ([]health.Reading, error) { return nil, errors.New("the bus did not answer") },
func() (map[string][]byte, error) { return backupHolders("hs", "restic"), nil })
s := m.Pass(context.Background(), nil)
said := strings.Join(append(append([]string{}, s.Said...), s.Warnings...), "\n")
if !strings.Contains(said, "2h0m0s") || !strings.Contains(said, "the bus did not answer") {
@@ -171,3 +171,21 @@ func TestAStaleOrUnknownBackupListIsSaidWithItsAge(t *testing.T) {
t.Errorf("the health does not say the backup list's age: %+v", h)
}
}
// backupHolders is the placement seat's state with one holder of node-backup on one node.
func backupHolders(node, module string) map[string][]byte {
return map[string][]byte{"node-backup." + node: []byte(`{"v":1,"seat":"node-backup","scope":"node","node":"` + node +
`","holders":[{"module":"` + module + `","node":"` + node + `"}]}`)}
}
func noHolders() (map[string][]byte, error) { return nil, errors.New("the state is not read yet") }
// readingsOfPaths is disk-load's readings of one node's items, by "<module>/<item>", each at its path.
func readingsOfPaths(node string, items map[string]string) []health.Reading {
var out []health.Reading
for k, p := range items {
module, item, _ := strings.Cut(k, "/")
out = append(out, health.Reading{V: 1, Node: node, PlacedDataItem: catalogue.PlacedDataItem{Module: module, ID: item, Path: p}})
}
return out
}
@@ -4,7 +4,7 @@ package main
import (
"context"
"encoding/json"
"git.novox.be/novox/mesh-sdk/go/health"
"strings"
"testing"
)
@@ -63,12 +63,10 @@ func TestAFailedReloadIsTriedAgainWhenNothingElseAsks(t *testing.T) {
}
}
func TestBackupFromRefusesAnAnswerWithoutADataList(t *testing.T) {
seats := json.RawMessage(`{"answer":{"seats":[{"seat":"node-backup","holders":[{"module":"restic","node":"ace"}]}]}}`)
for _, data := range []string{`{"answer":{}}`, `{"answer":{"data":null}}`} {
if _, paths, err := BackupFrom("ace", seats, json.RawMessage(data)); err == nil {
t.Errorf("%s was read as the empty list %v: never \"nothing to fear\"", data, paths)
}
func TestBackupFromRefusesNoReadingsAtAll(t *testing.T) {
seats := backupHolders("ace", "restic")
if _, paths, err := BackupFrom("ace", seats, []health.Reading{}); err == nil {
t.Errorf("no readings were read as the empty list %v: never \"nothing to fear\"", paths)
}
}
+47 -3
View File
@@ -16,10 +16,13 @@ import (
"os"
"os/signal"
"sort"
"strings"
"syscall"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
)
// ExportsState is the state the exports are kept in on the bus; MountsState the `mounts` modules' state,
@@ -88,9 +91,50 @@ func ShareExports(m *Machine) {
}
}
// askBus calls a seat's verb over the bus as this module, through the runtime.
func askBus(key string, body any) (json.RawMessage, error) {
return stdio.Ask(key, body)
// readingsOf reads disk-load's readings of one machine's data items, every key `<node>.<module>.<item>`.
func readingsOf(node string) ([]health.Reading, error) {
state := stdio.State(health.ReadingsRead)
names, err := state.Keys()
if err != nil {
return nil, err
}
var out []health.Reading
for _, k := range names {
if !strings.HasPrefix(k, node+".") {
continue
}
e, err := state.Get(k)
if err != nil {
return nil, err
}
if e == nil {
continue
}
if r, err := health.DecodeReading(e.Value); err == nil {
out = append(out, r)
}
}
return out, nil
}
// seatHolderKeys reads the placement seat's state of who holds which seat, every key.
func seatHolderKeys() (map[string][]byte, error) {
state := stdio.State(delivery.SeatHoldersRead)
names, err := state.Keys()
if err != nil {
return nil, err
}
keys := map[string][]byte{}
for _, k := range names {
e, err := state.Get(k)
if err != nil {
return nil, err
}
if e != nil {
keys[k] = e.Value
}
}
return keys, nil
}
// TakeWants reads every node's `mounts` state from the bus and writes what is asked of this machine to
@@ -2,6 +2,8 @@ package main
import (
"encoding/json"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
"os"
"regexp"
"sort"
@@ -137,7 +139,7 @@ func TestTheManifestCarriesGrantsTheRosterTheWantsAndTheRemovalOrder(t *testing.
if err := json.Unmarshal([]byte(raw), &full); err != nil {
t.Fatal(err)
}
if len(full.Reads) != 1 || full.Reads[0] != MountsState {
if len(full.Reads) != 3 || full.Reads[0] != health.ReadingsRead || full.Reads[1] != delivery.SeatHoldersRead || full.Reads[2] != MountsState {
t.Errorf("reads: %v", full.Reads)
}
roster := full.Facts["machines"]
+2 -2
View File
@@ -1,5 +1,5 @@
module nfsserver
go 1.22
go 1.26.0
require git.novox.be/novox/mesh-sdk/go v0.1.7
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029221656-196c5b6af277
+14 -2
View File
@@ -1,2 +1,14 @@
git.novox.be/novox/mesh-sdk/go v0.1.7 h1:C0sTQmtTiyYH7bnqZb7PusXnqA37gKuT7Nqjn9gG47w=
git.novox.be/novox/mesh-sdk/go v0.1.7/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029221656-196c5b6af277 h1:ajjsH4MWcx45MGYASjUSRQKhLzEringuk3TA0oxZRCU=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029221656-196c5b6af277/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
+2 -4
View File
@@ -48,12 +48,10 @@
}
],
"reads": [
"disk-load.readings",
"mesh-delivery-assigner.seat-holders",
"mounts.shares"
],
"invokes": [
"seat:mesh-controller.seats",
"seat:mesh-controller.data"
],
"config": {
"shares": {
"type": "string",
@@ -0,0 +1 @@
restic-backups
@@ -25,6 +25,7 @@ import (
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-sdk/go/health"
"os"
"os/exec"
"path/filepath"
@@ -32,6 +33,7 @@ import (
"slices"
"strconv"
"strings"
"sync"
"syscall"
"time"
)
@@ -188,6 +190,15 @@ type Backups struct {
// measuring run as processes of their own). Node is the machine's name, said in the events.
Emit func(event string, body any) error
Node string
// Readings is every reading disk-load keeps for this machine (its state `readings`, ADR 0571 clause 3); nil where the
// process has no runtime to read it.
Readings func() ([]health.Reading, error)
// PutGood writes one row of the state `good` (health.GoodRow): the night's process through its own bus account, the
// tools through the runtime; nil where there is no bus to write to.
PutGood func(key string, row health.GoodRow) error
goodMu sync.Mutex
goodWrote map[string][]byte
}
// The locks, files in this module's state. The night runs in a process of its own and `now`, `restore` and a
+40 -352
View File
@@ -1,26 +1,9 @@
// The data the modules on this machine declare, measured (novox/hq ADR 0233).
// The data the modules on this machine declare (novox/hq ADR 0233), and what the readings say of them.
//
// The mesh composes a second file beside the backup lines: every data item every module on the
// machine declares, one line each,
//
// item <id> <class> <path> <covered-by> <protection> <measure>
//
// where covered-by is the path whose snapshot keeps it (the item itself, or the directory its dump
// writes into), or `-` when it is not backed up; protection is backup, redundancy, both, or none; and
// measure is how the item is measured. This holder measures each item that is not a cache — its size,
// its newest write, and the redundant storage it is on and whether that is healthy (ZFS, md, btrfs) —
// and says it, with each module's last good backup and the precision of the measurement, in
// `backed-up`.
//
// **Never an unbounded walk of something large.** Three methods, the item's own choice:
//
// - `walk` (the default, for small items): every file summed and the newest change found, in one walk
// as root — at most once a day, and stopped after walkFor or walkFiles, said as "measured partially";
// - `dataset`: the ZFS dataset holding the path — its `used` from the filesystem's own counters — and
// the newest change among the path's top-level entries; hourly, and nothing is walked;
// - `shallow`: the newest change among the top-level entries only, and no size; hourly. The controller keeps the readings and says when an item shrinks, stops being written,
//
// goes without a backup, or is replaced by an empty copy of itself; this holder only measures.
// The mesh composes a file beside the backup lines: every data item every module on the machine declares, one line
// each, which mesh-sdk reads (catalogue.ParseDataItems). disk-load measures the items and keeps each reading and its
// seven-day peak as its own state (ADR 0571 clause 3); this holder reads them there and says each, with each module's
// last good backup, in `backed-up`.
//
// And it deletes, when a person has decided: an item the mesh retired — its module gone from this
// machine — is removed by `backup_delete_retired`, and only after a last restore point of it has been
@@ -32,104 +15,32 @@ import (
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/health"
"os"
"path/filepath"
"regexp"
"strconv"
"strings"
"sync"
"time"
)
// Item is one data item a module declares.
type Item struct {
// Measure is walk, dataset or shallow.
Measure string `json:"measure,omitempty"`
Module string `json:"-"`
ID string `json:"item"`
Class string `json:"class"`
Path string `json:"path"`
CoveredBy string `json:"covered_by,omitempty"`
Protection string `json:"protection,omitempty"`
}
// Item is one data item a module declares, as placed on this machine (mesh-sdk's reading of the holder's data text).
type Item = catalogue.PlacedDataItem
// Redundancy is the redundant storage an item is on, as read here.
type Redundancy struct {
// Kind is zfs, md or btrfs.
Kind string `json:"kind"`
// Where is the pool, the array device or the filesystem.
Where string `json:"where"`
// Healthy is nil when its state could not be read.
Healthy *bool `json:"healthy"`
Said string `json:"said"`
}
// Measured is what one measurement found.
type Measured struct {
SizeBytes *int64 `json:"size_bytes,omitempty"`
LastWrite *time.Time `json:"last_write,omitempty"`
MeasuredAt *time.Time `json:"measured_at,omitempty"`
// Precision says what the size is: "exact", "dataset <name>: its whole size", "partial: …" (a lower
// bound, never compared), or "no size: …".
Precision string `json:"precision,omitempty"`
Error string `json:"error,omitempty"`
Redundancy *Redundancy `json:"redundancy,omitempty"`
}
// ItemReport is one item as `backed-up` says it.
// ItemReport is one item as `backed-up` says it: the item, its newest reading from disk-load (ADR 0571 clause 3) and
// its newest good backup.
type ItemReport struct {
Item
Measured
health.Measured
// Peak is the largest size in the last seven days, on PeakDay.
Peak *int64 `json:"peak_bytes,omitempty"`
PeakDay string `json:"peak_day,omitempty"`
LastBackup *time.Time `json:"last_backup,omitempty"`
}
// The classes the mesh declares; a cache is listed and never measured.
const classCache = "cache"
var safePath = regexp.MustCompile(`^/[^\s'"\\$` + "`" + `]*$`)
// parseItems reads the composed data file into each module's items. A line this holder does not read
// is refused, naming the module, as a backup line is.
func parseItems(text string) (map[string][]Item, error) {
out := map[string][]Item{}
module := ""
for _, raw := range strings.Split(text, "\n") {
line := strings.TrimSpace(raw)
if line == "" {
continue
}
if m := moduleHeader.FindStringSubmatch(line); m != nil {
module = m[1]
continue
}
if strings.HasPrefix(line, "#") || module == "" {
continue
}
f := strings.Fields(line)
if len(f) < 5 || len(f) > 7 || f[0] != "item" || !safePath.MatchString(f[3]) || (f[4] != "-" && !safePath.MatchString(f[4])) {
return nil, fmt.Errorf("%s declares a data line this holder does not read: %s", module, line)
}
it := Item{Module: module, ID: f[1], Class: f[2], Path: f[3]}
if len(f) >= 6 {
it.Protection = f[5]
}
it.Measure = measureWalk
if len(f) == 7 {
it.Measure = f[6]
}
switch it.Measure {
case measureWalk, measureDataset, measureShallow:
default:
return nil, fmt.Errorf("%s declares a data line this holder does not read: %s", module, line)
}
if f[4] != "-" {
it.CoveredBy = f[4]
}
out[module] = append(out[module], it)
}
return out, nil
}
// Items is what the modules on this machine declare; none when the mesh composed no data file — an
// older controller, which composes none.
func (b *Backups) Items() (map[string][]Item, error) {
@@ -143,257 +54,30 @@ func (b *Backups) Items() (map[string][]Item, error) {
if err != nil {
return nil, err
}
return parseItems(string(raw))
return catalogue.ParseDataItems(string(raw))
}
func (b *Backups) measuredFile() string { return filepath.Join(b.Where.State, "measured.json") }
// Measurements is the newest measurement of each item, by module and item.
func (b *Backups) Measurements() map[string]map[string]Measured {
out := map[string]map[string]Measured{}
if raw, err := os.ReadFile(b.measuredFile()); err == nil {
_ = json.Unmarshal(raw, &out)
// Measurements is the newest reading disk-load keeps of each item on this machine, by module and item (its state
// `readings`, ADR 0571 clause 3); none where it cannot be read, which `backed-up` then says as no reading.
func (b *Backups) Measurements() map[string]map[string]health.Reading {
out := map[string]map[string]health.Reading{}
if b.Readings == nil {
return out
}
rs, err := b.Readings()
if err != nil {
b.Say("disk-load's readings could not be read: %v", err)
return out
}
for _, r := range rs {
if out[r.Module] == nil {
out[r.Module] = map[string]health.Reading{}
}
out[r.Module][r.ID] = r
}
return out
}
// The ways an item is measured.
const (
measureWalk = "walk"
measureDataset = "dataset"
measureShallow = "shallow"
)
// The bounds on a walk, and how often one runs: a walk is for small items, and one that meets a large
// one stops and says so rather than loading the disks for hours.
var (
walkFor = 10 * time.Minute
walkFiles = 2_000_000
walkEvery = 23 * time.Hour
)
// walkCommand sums the files under a path and finds its newest change, in one walk, as root — bounded
// by time and by files. One line out: "<bytes> <epoch seconds> <files> <complete 1|0>".
func walkCommand(path string) string {
return fmt.Sprintf("timeout %d find '%s' -xdev -printf '%%y %%s %%T@\\n' 2>/dev/null | head -n %d | "+
"awk 'BEGIN{s=0;m=0;n=0} {n++; if ($1==\"f\") s+=$2; if ($3>m) m=$3} END {printf \"%%d %%.0f %%d\\n\", s, m, n}'; "+
"echo \"${PIPESTATUS[0]:-0}\"", int(walkFor.Seconds()), path, walkFiles)
}
// topCommand is the newest change among a path and its top-level entries, and how many there are:
// one directory read, bounded. "<epoch seconds> <entries>".
func topCommand(path string) string {
return "timeout 60 find '" + path + "' -maxdepth 1 -printf '%T@\\n' 2>/dev/null | " +
"awk 'BEGIN{m=0;n=0} {n++; if ($1>m) m=$1} END {printf \"%.0f %d\\n\", m, n-1}'"
}
func epochTime(s string) *time.Time {
epoch, err := strconv.ParseInt(s, 10, 64)
if err != nil || epoch <= 0 {
return nil
}
t := time.Unix(epoch, 0).UTC()
return &t
}
// measure is one item, measured now, by its own method.
func (b *Backups) measure(ctx context.Context, it Item) Measured {
at := b.Now().UTC()
m := Measured{MeasuredAt: &at}
if !b.exists(ctx, it.Path) {
m.Error = it.Path + " does not exist"
zero := int64(0)
m.SizeBytes, m.Precision = &zero, "exact"
return m
}
m.Redundancy = b.redundancyOf(ctx, it.Path)
switch it.Measure {
case measureDataset, measureShallow:
out, err := b.Run(ctx, "bash", "-c", topCommand(it.Path))
if f := strings.Fields(out); err == nil && len(f) == 2 {
m.LastWrite = epochTime(f[0])
} else if err != nil {
m.Error = err.Error()
}
m.Precision = "no size: the newest change among its top-level entries only"
if it.Measure == measureShallow {
return m
}
out, err = b.Run(ctx, "zfs", "list", "-Hp", "-o", "name,used", it.Path)
f := strings.Fields(out)
if err != nil || len(f) != 2 {
m.Precision = "no size: it is on no ZFS dataset to read one from; the newest change among its top-level entries only"
return m
}
used, err := strconv.ParseInt(f[1], 10, 64)
if err != nil {
return m
}
m.SizeBytes = &used
m.Precision = "dataset " + f[0] + ": its whole size, from the filesystem's counters; the newest change among the top-level entries"
return m
}
out, err := b.Run(ctx, "bash", "-c", walkCommand(it.Path))
if err != nil {
m.Error = err.Error()
return m
}
lines := strings.Split(strings.TrimSpace(out), "\n")
f := strings.Fields(lines[0])
if len(f) != 3 || len(lines) < 2 {
m.Error = "the measurement said " + strings.TrimSpace(out)
return m
}
size, err1 := strconv.ParseInt(f[0], 10, 64)
files, err2 := strconv.Atoi(f[2])
if err1 != nil || err2 != nil {
m.Error = "the measurement said " + strings.TrimSpace(out)
return m
}
m.SizeBytes, m.LastWrite = &size, epochTime(f[1])
m.Precision = "exact"
if status := strings.TrimSpace(lines[len(lines)-1]); status == "124" || files >= walkFiles {
m.Precision = fmt.Sprintf("partial: measured partially — stopped after %s or %d files; the size is a lower bound, "+
"and this item wants measure: dataset or shallow", walkFor, walkFiles)
}
return m
}
// redundancyOf is the redundant storage a path is on, read as root: a ZFS pool's health and its own
// verdict, an md array's members, a btrfs filesystem's error counters. Nil for storage with no
// redundancy this holder knows how to read — which, for an item said to be protected by redundancy, the
// controller says is no protection at all.
func (b *Backups) redundancyOf(ctx context.Context, path string) *Redundancy {
out, err := b.Run(ctx, "findmnt", "-no", "SOURCE,FSTYPE", "--target", path)
if err != nil {
return nil
}
f := strings.Fields(out)
if len(f) < 2 {
return nil
}
source, fstype := f[0], f[1]
yes, no := true, false
switch {
case fstype == "zfs":
pool, _, _ := strings.Cut(source, "/")
r := &Redundancy{Kind: "zfs", Where: pool}
health, err := b.Run(ctx, "zpool", "list", "-H", "-o", "health", pool)
if err != nil {
r.Said = "zpool list: " + err.Error()
return r
}
status, err := b.Run(ctx, "zpool", "status", "-x", pool)
if err != nil {
r.Said = "zpool status: " + err.Error()
return r
}
health, status = strings.TrimSpace(health), strings.TrimSpace(status)
r.Said = "health " + health + "; " + firstLineOf(status)
if health == "ONLINE" && strings.Contains(status, "is healthy") {
r.Healthy = &yes
} else {
r.Healthy = &no
r.Said = "health " + health + "; " + oneLineOf(status)
}
return r
case strings.HasPrefix(source, "/dev/md"):
device := strings.TrimPrefix(source, "/dev/")
r := &Redundancy{Kind: "md", Where: device}
mdstat, err := b.Run(ctx, "cat", "/proc/mdstat")
if err != nil {
r.Said = err.Error()
return r
}
healthy, said, found := mdHealth(mdstat, device)
if !found {
r.Said = device + " is not in /proc/mdstat"
return r
}
r.Said = said
if healthy {
r.Healthy = &yes
} else {
r.Healthy = &no
}
return r
case fstype == "btrfs":
r := &Redundancy{Kind: "btrfs", Where: source}
stats, err := b.Run(ctx, "btrfs", "device", "stats", "--check", path)
if err != nil {
r.Healthy, r.Said = &no, "device errors: "+oneLineOf(err.Error())
return r
}
r.Healthy, r.Said = &yes, firstLineOf(stats)
return r
}
return nil
}
var mdMembers = regexp.MustCompile(`\[([U_]+)\]`)
// mdHealth reads one array's block of /proc/mdstat: healthy when every member is up and it is not
// recovering.
func mdHealth(mdstat, device string) (bool, string, bool) {
lines := strings.Split(mdstat, "\n")
for i, l := range lines {
if !strings.HasPrefix(l, device+" ") {
continue
}
block := l
for j := i + 1; j < len(lines) && strings.HasPrefix(lines[j], " "); j++ {
block += " " + strings.TrimSpace(lines[j])
}
members := mdMembers.FindStringSubmatch(block)
healthy := members != nil && !strings.Contains(members[1], "_") && !strings.Contains(block, "recovery")
return healthy, oneLineOf(block), true
}
return false, "", false
}
func firstLineOf(s string) string {
line, _, _ := strings.Cut(strings.TrimSpace(s), "\n")
return line
}
func oneLineOf(s string) string { return strings.Join(strings.Fields(s), " ") }
// MeasureAll measures every item that is not a cache, one at a time, and keeps what it found. 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.
func (b *Backups) MeasureAll(ctx context.Context, walks bool) error {
// The hourly round runs in its own process and the walk after a night in the night's (novox/hq issue 454).
release, err := b.hold(measureLock)
if err != nil {
return err
}
defer release()
items, err := b.Items()
if err != nil {
return err
}
before := b.Measurements()
found := map[string]map[string]Measured{}
for module, its := range items {
for _, it := range its {
if it.Class == classCache {
continue
}
if found[module] == nil {
found[module] = map[string]Measured{}
}
if was, ok := before[module][it.ID]; ok && it.Measure == measureWalk && !walks && was.MeasuredAt != nil &&
b.Now().Sub(*was.MeasuredAt) < walkEvery {
found[module][it.ID] = was
continue
}
found[module][it.ID] = b.measure(ctx, it)
}
}
raw, _ := json.MarshalIndent(found, "", " ")
return os.WriteFile(b.measuredFile(), append(raw, '\n'), 0o600)
}
func (b *Backups) goodFile() string { return filepath.Join(b.Where.State, "good.json") }
// Good is each module's newest good night: what a night that failed since does not erase.
@@ -418,14 +102,18 @@ func (b *Backups) recordGood(module string, at time.Time) {
if err := os.WriteFile(b.goodFile(), append(raw, '\n'), 0o600); err != nil {
b.Say("recording %s's good night failed: %v", module, err)
}
b.keepGood()
}
// itemReports is every item of one module, with its newest measurement and its newest good backup: the
// itemReports is every item of one module, with its newest reading and its newest good backup: the
// module's newest good night, where that night keeps the path that covers the item.
func itemReports(its []Item, measured map[string]Measured, paths []string, good time.Time) []ItemReport {
func itemReports(its []Item, measured map[string]health.Reading, paths []string, good time.Time) []ItemReport {
out := []ItemReport{}
for _, it := range its {
r := ItemReport{Item: it, Measured: measured[it.ID]}
r := ItemReport{Item: it}
if m, ok := measured[it.ID]; ok {
r.Measured, r.Peak, r.PeakDay = m.Measured, m.Peak, m.PeakAt
}
if it.CoveredBy != "" && !good.IsZero() {
for _, p := range paths {
if p == it.CoveredBy {
+33 -141
View File
@@ -6,7 +6,8 @@ package main
import (
"context"
"errors"
"fmt"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/health"
"os"
"path/filepath"
"strings"
@@ -28,53 +29,34 @@ func withData(t *testing.T, declared, data string) Where {
}
func TestTheComposedDataFileIsReadIntoEachModulesItems(t *testing.T) {
got, err := parseItems(composedData)
b := &Backups{Where: withData(t, composed, composedData), Now: time.Now, Say: quiet}
got, err := b.Items()
must(t, err)
if len(got["postgres"]) != 2 || got["postgres"][0].CoveredBy != "/var/lib/mesh-store/dumps" || got["postgres"][1].CoveredBy != "" ||
got["searxng"][0].Class != "cache" {
t.Fatalf("%+v", got)
}
for _, bad := range []string{"# pg\nitem a irreplaceable relative -\n", "# pg\nitem a irreplaceable /x\n", "# pg\nitem a b /x; rm -rf / -\n"} {
if _, err := parseItems(bad); err == nil || !strings.Contains(err.Error(), "pg declares a data line") {
t.Errorf("%q: %v", bad, err)
}
}
// An older controller composes no data file: nothing is measured, nothing fails.
b := &Backups{Where: placed(t, composed), Now: time.Now, Say: quiet}
// An older controller composes no data file: nothing is declared, nothing fails.
b = &Backups{Where: placed(t, composed), Now: time.Now, Say: quiet}
if items, err := b.Items(); err != nil || len(items) != 0 {
t.Fatalf("%v, %v", items, err)
}
}
// Every item but a cache is measured — its files' size and its newest change, in one walk as root — and
// `backed-up` says each with the newest good night of the module that keeps the path covering it.
func TestEveryItemButACacheIsMeasuredAndSaidWithItsLastGoodBackup(t *testing.T) {
// `backed-up` says every item with disk-load's newest reading and its peak, and the newest good night of the module that
// keeps the path covering it; an item with no reading yet is said with none.
func TestEveryItemIsSaidWithItsReadingAndItsLastGoodBackup(t *testing.T) {
where := withData(t, composed, composedData)
var mu sync.Mutex
var walked []string
run := func(_ context.Context, name string, args ...string) (string, error) {
mu.Lock()
defer mu.Unlock()
switch {
case name == "test":
return "", nil
case name == "bash" && strings.Contains(args[1], "find '"):
walked = append(walked, args[1])
if strings.Contains(args[1], "'/var/lib/mesh-store'") {
return "1073741824 1759744800 4000\n0\n", nil
}
return "5000 1759700000 3\n0\n", nil
case name == "restic":
return "[]", nil
}
return "", nil
}
size, peak := int64(1<<30), int64(2<<30)
at := time.Date(2026, 10, 6, 4, 0, 0, 0, time.UTC)
write := time.Unix(1759744800, 0).UTC()
night := time.Date(2026, 10, 6, 3, 0, 0, 0, time.UTC)
b := &Backups{Where: where, Run: run, Now: func() time.Time { return night.Add(time.Hour) }, Say: quiet}
must(t, b.MeasureAll(context.Background(), true))
if len(walked) != 2 {
t.Fatalf("walked %d paths, want the two that are not a cache: %v", len(walked), walked)
}
b := &Backups{Where: where, Now: func() time.Time { return night.Add(time.Hour) }, Say: quiet,
Run: func(context.Context, string, ...string) (string, error) { return "[]", nil },
Readings: func() ([]health.Reading, error) {
return []health.Reading{{V: 1, Node: "ace", PlacedDataItem: catalogue.PlacedDataItem{Module: "postgres", ID: "store"},
Measured: health.Measured{SizeBytes: &size, LastWrite: &write, MeasuredAt: &at, Precision: "exact"}, Peak: &peak, PeakAt: "2026-10-05"}}, nil
}}
b.recordGood("postgres", night)
got, err := b.BackedUp(context.Background(), "")
must(t, err)
@@ -89,12 +71,23 @@ func TestEveryItemButACacheIsMeasuredAndSaidWithItsLastGoodBackup(t *testing.T)
}
}
}
if store == nil || store.SizeBytes == nil || *store.SizeBytes != 1<<30 || store.LastWrite == nil ||
store.LastWrite.Unix() != 1759744800 || store.LastBackup == nil || !store.LastBackup.Equal(night) {
if store == nil || store.SizeBytes == nil || *store.SizeBytes != 1<<30 || store.Peak == nil || *store.Peak != 2<<30 ||
store.LastWrite == nil || store.LastWrite.Unix() != 1759744800 || store.LastBackup == nil || !store.LastBackup.Equal(night) {
t.Fatalf("the store is said as %+v", store)
}
if dumps == nil || dumps.LastBackup != nil {
t.Fatalf("an item not backed up is said backed up: %+v", dumps)
if dumps == nil || dumps.LastBackup != nil || dumps.SizeBytes != nil {
t.Fatalf("an item with no reading or no backup is said with one: %+v", dumps)
}
// The same last good night is kept on the bus beside good.json, one row per item a backup path covers.
b.Node = "ace"
wrote := map[string]health.GoodRow{}
b.PutGood = func(k string, r health.GoodRow) error { wrote[k] = r; return nil }
b.recordGood("postgres", night) // a good night writes the rows, once; the same night again writes nothing
n := len(wrote)
b.recordGood("postgres", night)
row, ok := wrote[health.GoodKey("ace", "postgres", "store")]
if len(wrote) != 1 || n != 1 || !ok || !row.LastGood.Equal(night) || row.CoveredBy != "/var/lib/mesh-store/dumps" || row.V != health.GoodVersion {
t.Fatalf("the good rows are %+v", wrote)
}
}
@@ -202,104 +195,3 @@ func TestARetiredItemIsDeletedOnlyAfterALastRestorePoint(t *testing.T) {
t.Fatalf("%+v, %v", again, err)
}
}
// The redundant storage under an item is read: a ZFS pool healthy, then degraded; an md array with a
// member missing; and plain storage, which is no redundancy at all.
func TestTheRedundantStorageUnderAnItemIsRead(t *testing.T) {
health, statusX := "ONLINE\n", "pool 'storage' is healthy\n"
fs := "storage/media zfs\n"
run := func(_ context.Context, name string, args ...string) (string, error) {
switch name {
case "findmnt":
return fs, nil
case "zpool":
if args[0] == "list" {
return health, nil
}
return statusX, nil
case "cat":
return "Personalities : [raid1]\nmd127 : active raid1 sdb1[1] sda1[0]\n 976630464 blocks super 1.2 [2/1] [U_]\n\nunused devices: <none>\n", nil
}
return "", nil
}
b := &Backups{Run: run, Now: time.Now, Say: quiet}
r := b.redundancyOf(context.Background(), "/storage/media/movies")
if r == nil || r.Kind != "zfs" || r.Where != "storage" || r.Healthy == nil || !*r.Healthy {
t.Fatalf("a healthy pool: %+v", r)
}
health, statusX = "DEGRADED\n", " pool: storage\n state: DEGRADED\nstatus: One or more devices has been removed\n"
if r := b.redundancyOf(context.Background(), "/storage/media/movies"); r.Healthy == nil || *r.Healthy || !strings.Contains(r.Said, "DEGRADED") {
t.Fatalf("a degraded pool: %+v", r)
}
fs = "/dev/md127 ext4\n"
if r := b.redundancyOf(context.Background(), "/srv/x"); r == nil || r.Kind != "md" || r.Healthy == nil || *r.Healthy {
t.Fatalf("an array missing a member: %+v", r)
}
fs = "/dev/nvme0n1p3 ext4\n"
if r := b.redundancyOf(context.Background(), "/var/lib/x"); r != nil {
t.Fatalf("plain storage read as redundant: %+v", r)
}
}
// Large items are never walked: a dataset is measured from the filesystem's counters and its top-level
// entries, a shallow item from its top-level entries only; a walk that meets more than its bound stops
// and says so; and a walk runs at most once a day — the hourly round keeps the last one.
func TestNothingLargeIsWalkedAndAWalkIsBoundedAndDaily(t *testing.T) {
data := "# media\nitem movies irreplaceable /storage/media/movies - redundancy dataset\n" +
"item meta rebuildable /mnt/plex/config /mnt/plex/config backup shallow\n" +
"item big valuable /srv/big /srv/big backup\n"
where := withData(t, composed, data)
var mu sync.Mutex
var walks []string
truncated := true
run := func(_ context.Context, name string, args ...string) (string, error) {
mu.Lock()
defer mu.Unlock()
line := name + " " + strings.Join(args, " ")
switch {
case name == "test":
return "", nil
case name == "zfs" && args[len(args)-1] == "/storage/media/movies":
return "storage/media\t97067690722560\n", nil
case name == "zfs":
return "", errors.New("not a ZFS dataset")
case name == "bash" && strings.Contains(line, "-maxdepth 1"):
return "1759744800 12673\n", nil
case name == "bash":
walks = append(walks, line)
if strings.Contains(line, "-xdev") && !strings.Contains(line, "'/srv/big'") {
t.Errorf("walked something declared not to be walked: %s", line)
}
if truncated {
return fmt.Sprintf("123 1759700000 %d\n141\n", walkFiles), nil
}
return "999 1759700000 10\n0\n", nil
}
return "", nil
}
now := time.Date(2026, 10, 6, 12, 0, 0, 0, time.UTC)
b := &Backups{Where: where, Run: run, Now: func() time.Time { return now }, Say: quiet}
must(t, b.MeasureAll(context.Background(), false))
got := b.Measurements()["media"]
if m := got["movies"]; m.SizeBytes == nil || *m.SizeBytes != 97067690722560 || !strings.HasPrefix(m.Precision, "dataset storage/media") || m.LastWrite == nil {
t.Fatalf("the library from its dataset: %+v", m)
}
if m := got["meta"]; m.SizeBytes != nil || !strings.HasPrefix(m.Precision, "no size") || m.LastWrite == nil {
t.Fatalf("a shallow item: %+v", m)
}
if m := got["big"]; !strings.HasPrefix(m.Precision, "partial") {
t.Fatalf("a walk stopped at its bound: %+v", m)
}
// An hour later the hourly round reads counters again and walks nothing.
now = now.Add(time.Hour)
truncated = false
must(t, b.MeasureAll(context.Background(), false))
if len(walks) != 1 {
t.Fatalf("walked %d times in two rounds an hour apart: %v", len(walks), walks)
}
// After the night, it walks.
must(t, b.MeasureAll(context.Background(), true))
if len(walks) != 2 || b.Measurements()["media"]["big"].Precision != "exact" {
t.Fatalf("%d walks, %+v", len(walks), b.Measurements()["media"]["big"])
}
}
+128
View File
@@ -0,0 +1,128 @@
package main
// **Each item's last good night, on the bus** (novox/hq ADR 0571 clause 3, ADR 0564 §6): beside good.json, the module's
// own state `good` (health.GoodRow), one key `<node>.<module>.<item>`, written when a night's files change, so the
// self-check's probe of a stale backup reads it there.
import (
"encoding/json"
"path/filepath"
"strings"
"syscall"
"time"
"unsafe"
"git.novox.be/novox/mesh-sdk/go/health"
)
// GoodRows is the rows of this machine (health.GoodRow): every item a module declares that a path of the module's backup
// covers, with the module's newest good night.
func (b *Backups) GoodRows() (map[string]health.GoodRow, error) {
items, err := b.Items()
if err != nil {
return nil, err
}
declared, err := b.Declared()
if err != nil {
return nil, err
}
paths := map[string][]string{}
for _, d := range declared {
paths[d.Module] = d.Paths
}
good := b.Good()
out := map[string]health.GoodRow{}
for module, its := range items {
for _, it := range its {
if it.Class == "cache" || it.CoveredBy == "" {
continue
}
covered := false
for _, p := range paths[module] {
covered = covered || p == it.CoveredBy
}
if !covered {
continue
}
out[health.GoodKey(b.Node, module, it.ID)] = health.GoodRow{V: health.GoodVersion, Node: b.Node, Module: module,
Item: it.ID, CoveredBy: it.CoveredBy, LastGood: good[module].UTC()}
}
}
return out, nil
}
// keepGood writes the rows that differ from what was last written, when a night succeeded (recordGood). A bus that
// cannot be reached loses nothing: good.json is the local copy, and the next good night writes the rows again.
func (b *Backups) keepGood() {
if b.PutGood == nil {
return
}
rows, err := b.GoodRows()
if err != nil {
b.Say("the last good nights could not be read: %v", err)
return
}
b.goodMu.Lock()
defer b.goodMu.Unlock()
if b.goodWrote == nil {
b.goodWrote = map[string][]byte{}
}
for k, r := range rows {
raw, _ := json.Marshal(r)
if string(b.goodWrote[k]) == string(raw) {
continue
}
if err := b.PutGood(k, r); err != nil {
b.Say("the last good night of %s was not put on the bus: %v", k, err)
continue
}
b.goodWrote[k] = raw
}
}
// watchGood writes the rows once at start and then each time the night's process changes what it keeps in the state
// directory (good.json, the nights' file), as the file system says it: the night runs as the operator account, which a
// process with a bus account of its own may not (ADR 0436 §2), so the tools' runtime, which has the module's account,
// writes at the night's event instead of the night itself. No timer.
func watchGood(b *Backups) {
time.Sleep(20 * time.Second) // the runtime serves first
b.keepGood()
for {
err := whenWritten(b.Where.State, func(name string) bool { return name == "good.json" || name == filepath.Base(b.nightsFile()) }, b.keepGood)
b.Say("watching the state directory for a night failed (%v); trying again in a minute", err)
time.Sleep(time.Minute)
b.keepGood()
}
}
// whenWritten calls then each time a file of dir that match names is written or put in place, until the watch fails.
func whenWritten(dir string, match func(name string) bool, then func()) error {
fd, err := syscall.InotifyInit1(syscall.IN_CLOEXEC)
if err != nil {
return err
}
defer syscall.Close(fd)
if _, err := syscall.InotifyAddWatch(fd, dir, syscall.IN_CLOSE_WRITE|syscall.IN_MOVED_TO); err != nil {
return err
}
buf := make([]byte, 64*syscall.SizeofInotifyEvent+4096)
for {
n, err := syscall.Read(fd, buf)
if err != nil {
if err == syscall.EINTR {
continue
}
return err
}
heard := false
for off := 0; off+syscall.SizeofInotifyEvent <= n; {
ev := (*syscall.InotifyEvent)(unsafe.Pointer(&buf[off]))
name := strings.TrimRight(string(buf[off+syscall.SizeofInotifyEvent:off+syscall.SizeofInotifyEvent+int(ev.Len)]), "\x00")
heard = heard || match(name)
off += syscall.SizeofInotifyEvent + int(ev.Len)
}
if heard {
then()
}
}
}
@@ -263,8 +263,8 @@ func TestTheManifestRunsTheNightJudgedByItsHealthTool(t *testing.T) {
}
env := m.Build.Artifacts[0].Env
want := map[string]struct{ run, kind, tool string }{
"night": {"night", "tool", "backup_health"},
"measure": {"measure", "unit", ""},
"night": {"night", "unit", ""},
"awake": {"awake", "tool", "backup_health"},
}
for _, r := range m.Resources {
w, ok := want[r["id"].(string)]
+40 -25
View File
@@ -24,6 +24,8 @@ import (
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/announce"
"git.novox.be/novox/mesh-sdk/go/health"
)
// Seat is the role this module holds.
@@ -46,20 +48,29 @@ func main() {
}
if way == "" {
b.Emit = func(event string, body any) error { return stdio.Emit(event, body) }
b.Readings = busReadings(b.Node)
}
switch way {
case "night":
nights(b)
case "measure":
measurements(b)
case "awake":
counting(b)
case "":
if err := stdio.Keeps(announce.State{Name: health.GoodState}); err != nil {
say("%v", err)
os.Exit(1)
}
b.PutGood = func(key string, row health.GoodRow) error {
_, err := stdio.State(health.GoodState).Put(key, row)
return err
}
go watchGood(b)
if err := stdio.Serve("", tools(b)); err != nil {
say("%v", err)
os.Exit(1)
}
default:
say("%q is not a way to run; with no argument it serves its tools, with night it runs the nights, with "+
"measure it measures the data declared here", way)
say("%q is not a way to run; with no argument it serves its tools, with night it runs the nights, with awake it counts how long the machine is up since a night", way)
os.Exit(2)
}
}
@@ -72,7 +83,6 @@ func nights(b *Backups) {
if h, err := strconv.Atoi(os.Getenv("MESH_BACKUP_HOUR")); err == nil && h >= 0 && h < 24 {
hour = h
}
go counting(b)
if missedANight(b.Nights(), time.Now()) {
time.Sleep(10 * time.Minute)
night(b)
@@ -83,23 +93,6 @@ func nights(b *Backups) {
}
}
// measurements measures every declared item once an hour (MESH_BACKUP_MEASURE_EVERY to change it),
// the first a few minutes after start — counters hourly, a walk at most daily — and every walk after
// each night (novox/hq ADR 0233).
func measurements(b *Backups) {
every := time.Hour
if d, err := time.ParseDuration(os.Getenv("MESH_BACKUP_MEASURE_EVERY")); err == nil && d >= time.Minute {
every = d
}
time.Sleep(3 * time.Minute)
for {
if err := b.MeasureAll(context.Background(), false); err != nil {
say("measuring the data declared here failed: %v", err)
}
time.Sleep(every)
}
}
func night(b *Backups) {
ctx := context.Background()
outcome, err := b.BackUp(ctx, "")
@@ -116,9 +109,6 @@ func night(b *Backups) {
if len(failed) > 0 {
say("the night left %s without a backup", strings.Join(failed, ", "))
}
if err := b.MeasureAll(ctx, true); err != nil {
say("measuring the data declared here failed: %v", err)
}
// Sundays, the repository's own integrity with a sample of the data read back.
if time.Now().Weekday() == time.Sunday {
if err := b.Check(ctx); err != nil {
@@ -204,3 +194,28 @@ func tools(b *Backups) []stdio.Tool {
Run: func(a map[string]any) (any, error) { return b.DeletedOutcome(arg(a, "path")) }},
}
}
// busReadings reads disk-load's readings of this machine from its state, every key `<node>.<module>.<item>`.
func busReadings(node string) func() ([]health.Reading, error) {
return func() ([]health.Reading, error) {
st := stdio.State(health.ReadingsRead)
keys, err := st.Keys()
if err != nil {
return nil, err
}
var out []health.Reading
for _, k := range keys {
if !strings.HasPrefix(k, node+".") {
continue
}
e, err := st.Get(k)
if err != nil || e == nil {
continue
}
if r, err := health.DecodeReading(e.Value); err == nil {
out = append(out, r)
}
}
return out, nil
}
}
+11 -2
View File
@@ -1,5 +1,14 @@
module restic
go 1.25.0
go 1.26.0
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261010191001-33917f91ac3c
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029223737-9f0bb1cfa4ba
require (
github.com/klauspost/compress v1.20.0 // indirect
github.com/nats-io/nats.go v1.54.0 // indirect
github.com/nats-io/nkeys v0.4.16 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
golang.org/x/crypto v0.57.0 // indirect
golang.org/x/sys v0.48.0 // indirect
)
+26 -2
View File
@@ -1,2 +1,26 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261010191001-33917f91ac3c h1:l5onJwoIeH8yE/PjVlEeoPyXBsHp2NQiWLllz2fwX7s=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261010191001-33917f91ac3c/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029223737-9f0bb1cfa4ba h1:4By9fDPDDiJXX32ySd6+WjAMAIX9vkVTT+Zs9+vhcLk=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029223737-9f0bb1cfa4ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
+50 -44
View File
@@ -41,40 +41,40 @@
}
]
},
"config": {
"keep-daily": {
"type": "number",
"description": "How many nightly restore points are kept, a count from 1 to 9999: two weeks undo a mistake noticed within a fortnight. Read by the night, which thins nothing while it does not read (novox/hq issue 416); lowering it deletes the restore points beyond the new count at the next night, and that cannot be undone.",
"minimum": 1,
"maximum": 9999,
"trusted": true,
"default": 14
},
"keep-weekly": {
"type": "number",
"description": "How many weekly restore points are kept behind the dailies, a count from 1 to 9999, refused by the night like keep-daily; lowering it deletes the restore points beyond the new count at the next night, and that cannot be undone.",
"minimum": 1,
"maximum": 9999,
"trusted": true,
"default": 8
},
"keep-monthly": {
"type": "number",
"description": "How many monthly restore points are kept behind the weeklies, a count from 1 to 9999, refused by the night like keep-daily; lowering it deletes the restore points beyond the new count at the next night, and that cannot be undone.",
"minimum": 1,
"maximum": 9999,
"trusted": true,
"default": 6
},
"keep-hand-taken-days": {
"type": "number",
"description": "How many days a restore point taken by hand (node-backup.now, or the mesh's before a module moves) is kept at least, a count from 1 to 9999; it is kept until a later good night of its module exists, and this many days in any case, so the next thinning never takes the one taken against a risky change (novox/hq issue 453). Lowering it lets go of the older ones taken by hand at the next thinning, and that cannot be undone.",
"minimum": 1,
"maximum": 9999,
"trusted": true,
"default": 7
}
"config": {
"keep-daily": {
"type": "number",
"description": "How many nightly restore points are kept, a count from 1 to 9999: two weeks undo a mistake noticed within a fortnight. Read by the night, which thins nothing while it does not read (novox/hq issue 416); lowering it deletes the restore points beyond the new count at the next night, and that cannot be undone.",
"minimum": 1,
"maximum": 9999,
"trusted": true,
"default": 14
},
"keep-weekly": {
"type": "number",
"description": "How many weekly restore points are kept behind the dailies, a count from 1 to 9999, refused by the night like keep-daily; lowering it deletes the restore points beyond the new count at the next night, and that cannot be undone.",
"minimum": 1,
"maximum": 9999,
"trusted": true,
"default": 8
},
"keep-monthly": {
"type": "number",
"description": "How many monthly restore points are kept behind the weeklies, a count from 1 to 9999, refused by the night like keep-daily; lowering it deletes the restore points beyond the new count at the next night, and that cannot be undone.",
"minimum": 1,
"maximum": 9999,
"trusted": true,
"default": 6
},
"keep-hand-taken-days": {
"type": "number",
"description": "How many days a restore point taken by hand (node-backup.now, or the mesh's before a module moves) is kept at least, a count from 1 to 9999; it is kept until a later good night of its module exists, and this many days in any case, so the next thinning never takes the one taken against a risky change (novox/hq issue 453). Lowering it lets go of the older ones taken by hand at the next thinning, and that cannot be undone.",
"minimum": 1,
"maximum": 9999,
"trusted": true,
"default": 7
}
},
"resources": [
{
"id": "state",
@@ -139,22 +139,17 @@
"MESH_BACKUP_DATA": "${dir:state}/data.conf"
},
"health": {
"kind": "tool",
"tool": "backup_health",
"interval": "60s",
"timeout": "20s",
"looks": 2,
"grace": "60s"
"kind": "unit"
}
},
{
"id": "measure",
"id": "awake",
"type": "process",
"name": "restic-measure",
"name": "restic-awake",
"artifact": "tools",
"run": [
"./restic-backups",
"measure"
"awake"
],
"user": "${machine:account}",
"env": {
@@ -165,7 +160,12 @@
"MESH_BACKUP_DATA": "${dir:state}/data.conf"
},
"health": {
"kind": "unit"
"kind": "tool",
"tool": "backup_health",
"interval": "60s",
"timeout": "20s",
"looks": 2,
"grace": "60s"
}
}
],
@@ -195,5 +195,11 @@
"from": "files"
}
]
}
},
"reads": [
"disk-load.readings"
],
"state": [
"good"
]
}