Keep libraries under groups of their own, and give modules users in them (hq ADR 0328, issue 567)

ADR 0328 supersedes ADR 0325: shared data is a library with one home, kept usable by
its group rather than by ACL entries. The engine now makes each library's group and
each module's user in its libraries' groups; keeps a library's root and requested
sub-paths in its group with group write and setgid on its home, making missing
sub-paths as the root is; waits for the operator's word before touching a root that
holds data, and on the take act walks it by descriptor (no link, same filesystem,
bounded and resumed, files with other names left); holds the containers of a library
whose disk or mount is not there; and puts the tool runner's account in the group.
The ACL keeping, mask repair and watch are gone; an access is again as before ADR 0325.
This commit is contained in:
2026-10-12 22:11:09 +02:00
parent 303d4037e8
commit 2a2be687f3
35 changed files with 2583 additions and 5004 deletions
-439
View File
@@ -1,439 +0,0 @@
package main
import (
"context"
"errors"
"fmt"
"os"
"path/filepath"
"sort"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/novox/mesh-host/internal/apply"
"github.com/novox/mesh-host/internal/declaration"
"github.com/novox/mesh-host/internal/link"
"github.com/novox/mesh-host/internal/store"
)
// An access is judged by what each of its users may do (novox/hq ADR 0325 §7, §10, §12).
//
// On every look the engine judges each module's access as its module's resource of kind `access`: waiting, with a
// wait of kind access, while the mesh may not keep its folder yet or its disk is not there; unhealthy for root's
// data on its placed filesystem, and for a kept folder that gives a user less than its right — read by entry and
// mask together, on the folder and the sweep's sample under it, which the sweep puts back where it can; healthy
// otherwise. The folder is the target, never in a reason nor in a wait's words: the controller says them on the
// operator's channel, which withholds a path (ADR 0234 §6).
// judgedAccess is one access as the judge reads it.
type judgedAccess struct {
module, id, name, path string
taken bool
users []store.KeptUser
// leftover is the mesh's entries nobody wants, not an access (link.KindKeptEntries).
leftover bool
}
// accessJudge reads every access on every look. Safe for the apply and the looking loop at once.
type accessJudge struct {
mu sync.Mutex
accesses []judgedAccess
// was is each access's last verdict, with since when and how many looks in a row found it unhealthy.
was map[string]link.ResourceHealth
// reading guards inFlight: every folder whose look has not returned yet.
reading sync.Mutex
inFlight map[string]bool
}
// accesses is the one access judge on this machine.
var accesses = newAccessJudge()
func newAccessJudge() *accessJudge {
return &accessJudge{was: map[string]link.ResourceHealth{}, inFlight: map[string]bool{}}
}
// set is what the declaration just applied asks: every access of a module, not held as found, with the users of
// its folder as every access of that folder gives them. An access no longer declared is forgotten.
func (j *accessJudge) set(d *declaration.Declaration, held map[string]bool) {
var as []judgedAccess
if d != nil {
// A tool runner's account that came with no number is read by name, as the apply read it (ADR 0325 §4).
ctx, cancel := context.WithTimeout(context.Background(), lookBound)
apply.ResolveRunners(ctx, d)
cancel()
users := map[string]map[uint32]store.KeptUser{}
taken := map[string]bool{}
for _, r := range d.Resources {
if a, ok := r.(*declaration.Access); ok {
path := filepath.Clean(a.Path)
if users[path] == nil {
users[path] = map[uint32]store.KeptUser{}
}
taken[path] = taken[path] || a.Taken
for _, u := range a.Users {
had, seen := users[path][uint32(u.UID)]
if !seen || (u.Use == declaration.AccessReadWrite && had.Use != declaration.AccessReadWrite) {
users[path][uint32(u.UID)] = store.KeptUser{Identity: u.Identity, UID: uint32(u.UID),
Use: u.Use, From: u.From}
}
}
}
}
for _, r := range d.Resources {
a, ok := r.(*declaration.Access)
if !ok || held[a.ID] || strings.HasPrefix(a.ID, declaration.AdoptionPrefix) {
continue
}
// An older controller's access is judged by nothing, as before ADR 0325: no state is said of it.
if !apply.Keeps(d, a.Path) {
continue
}
at := strings.LastIndex(a.ID, ".")
if at <= 0 {
continue
}
path := filepath.Clean(a.Path)
ja := judgedAccess{module: a.ID[:at], id: a.ID, name: a.ID[at+1:], path: path, taken: taken[path]}
for _, u := range users[path] {
ja.users = append(ja.users, u)
}
sort.Slice(ja.users, func(x, y int) bool { return ja.users[x].UID < ja.users[y].UID })
as = append(as, ja)
}
}
sort.Slice(as, func(a, b int) bool { return as[a].id < as[b].id })
j.mu.Lock()
defer j.mu.Unlock()
j.accesses = as
declared := map[string]bool{}
for _, a := range as {
declared[a.id] = true
}
for id := range j.was {
if !declared[id] {
delete(j.was, id)
}
}
}
// lookBound is how long a look waits for one accessed folder to answer. A hard network mount whose server is down
// blocks a stat for ever, and one access must never stop every health statement this machine makes.
var lookBound = 5 * time.Second
// verdict is one access's judgement in a look.
type verdict struct {
state, reason string
waits []link.AccessWait
act *link.AccessActRecord
entries []link.KeptEntry
}
// judgeBounded judges one access within lookBound, outside any lock. A look that has not answered is said so;
// while it still hangs, later looks start no other on the same folder, so a dead mount costs one blocked read.
func (j *accessJudge) judgeBounded(a judgedAccess) (verdict, bool) {
j.reading.Lock()
if j.inFlight[a.path] {
j.reading.Unlock()
return verdict{}, false
}
j.inFlight[a.path] = true
j.reading.Unlock()
done := make(chan verdict, 1)
go func() {
v := judgeAccess(a)
j.reading.Lock()
delete(j.inFlight, a.path)
j.reading.Unlock()
done <- v
}()
select {
case v := <-done:
return v, true
case <-time.After(lookBound):
return verdict{}, false
}
}
// judgeAccess is one access's judgement now (ADR 0325 §7, §10, §12), the sweep of its kept folder included.
func judgeAccess(a judgedAccess) verdict {
v := apply.AccessVerdict(context.Background(), a.path, a.taken, len(a.users) > 0)
switch v.Kind {
case apply.VerdictWaitKeep:
return verdict{state: link.StateWaiting, reason: "it holds data the mesh has not taken on the filesystem " +
"it is on now, and waits for the operator's word (novox/hq ADR 0325 §7, §8)",
waits: []link.AccessWait{{Kind: link.WaitKindAccess, Access: a.name, Part: a.module,
What: apply.KeepWords(a.name), Placed: v.Placed, Reason: link.WaitKeep, ACL: v.ACL}}}
case apply.VerdictWaitDisk:
return verdict{state: link.StateWaiting, reason: "the disk that holds it is not there (novox/hq ADR 0325 §9)",
waits: []link.AccessWait{{Kind: link.WaitKindAccess, Access: a.name, Part: a.module,
What: apply.DiskWords(a.name), Reason: link.WaitDisk}}}
case apply.VerdictRoot:
return verdict{state: link.StateUnhealthy, reason: apply.RootWords(v.Why)}
case apply.VerdictUnknown:
// Not a fault and not the operator's: said as not known, and judged again on the next look.
return verdict{state: link.StateUnknown, reason: v.Why + "; judged again on the next look"}
case apply.VerdictBroken:
// Its words carry no path (AccessVerdict says them so): the reason reaches the operator's channel.
return verdict{state: link.StateUnhealthy, reason: v.Why}
case apply.VerdictMake:
return verdict{state: link.StateUnhealthy, reason: "it is not there yet; the next apply makes it where " +
"its data lives (novox/hq ADR 0325 §9)"}
case apply.VerdictNobody:
return verdict{state: link.StateHealthy}
}
out := verdict{state: link.StateHealthy, entries: entriesOf(a.users)}
rec, _ := keptNow(a.path)
if act := rec.Act; act != nil {
out.act = &link.AccessActRecord{Act: act.Act, Access: a.id, Warrant: act.Warrant, By: act.By, State: act.State,
Started: act.Started, Ended: act.Ended, Folders: act.Folders, Files: act.Files, Kept: act.Kept,
Links: act.Links, Other: act.Other, Mounts: act.Mounts, Why: act.Why}
}
if !rec.Kept {
out.state, out.reason = link.StateStarting, "the controller says it was taken; the next apply keeps it"
return out
}
look := apply.LookKept(a.path, a.users, apply.SweepLimits)
var faults []string
if look.Failed != "" {
faults = append(faults, "the sweep could not look: "+apply.Pathless(errors.New(look.Failed)))
}
if len(look.Short) > 0 {
faults = append(faults, "the folder itself gives "+strings.Join(look.Short, ", ")+"; put back on the next "+
"apply")
}
if n := look.Sweep.Unrepaired + look.Sweep.LastUnrepaired; n > 0 {
faults = append(faults, fmt.Sprintf("%d entries under it give a user less than its right and cannot be "+
"put back: an owner's own bits, which the mesh never changes, or an entry the filesystem refused", n))
}
if len(faults) > 0 {
out.state, out.reason = link.StateUnhealthy, strings.Join(faults, "; ")
return out
}
switch {
case rec.Act != nil && rec.Act.State == link.ActRunning:
out.state, out.reason = link.StateStarting, "the take act is adding its users' entries under it"
case !sameUsersList(rec.Users, rec.Want):
out.state, out.reason = link.StateStarting, "the entries under it are being changed to its users now"
}
return out
}
// entriesOf is a kept folder's listing (ADR 0325 §4): each user, and beside the tool runner's what it stands for.
func entriesOf(users []store.KeptUser) []link.KeptEntry {
var out []link.KeptEntry
for _, u := range users {
e := link.KeptEntry{Identity: u.Identity, UID: int(u.UID), Use: u.Use, From: u.From}
if u.From == declaration.FromToolRunner {
e.Note = "the tool runner's account stands for every module whose tools use this folder, at the widest " +
"use any of them declares: a module that declared only read may write through it (novox/hq ADR " +
"0325 §4, until issue 569)"
}
out = append(out, e)
}
return out
}
func sameUsersList(a, b []store.KeptUser) bool {
if len(a) != len(b) {
return false
}
seen := map[uint32]string{}
for _, u := range a {
seen[u.UID] = u.Use
}
for _, u := range b {
if use, ok := seen[u.UID]; !ok || use != u.Use {
return false
}
}
return true
}
// keptNow is the record of one kept folder, as the engine's file holds it.
func keptNow(path string) (store.KeptFolder, bool) {
all, err := apply.KeptFolders.All()
if err != nil {
return store.KeptFolder{}, false
}
f, had := all[path]
return f, had
}
// look reads every access now. Whether any verdict changed since the last look, for the loop to say it.
func (j *accessJudge) look(now time.Time) ([]link.ResourceHealth, bool) {
j.mu.Lock()
as := append([]judgedAccess(nil), j.accesses...)
j.mu.Unlock()
read := make([]verdict, len(as))
for i, a := range as {
v, answered := j.judgeBounded(a)
if !answered {
v = verdict{state: link.StateUnhealthy, reason: "its folder does not answer: judging it did not return " +
"within " + lookBound.String()}
}
read[i] = v
}
// And the entries nobody wants that could not be taken out yet, each under the access that last named its
// folder, until its drop walk has run (ADR 0325 §8).
for _, l := range apply.Leftovers() {
as = append(as, judgedAccess{module: l.Module, id: l.Access + "~entries", name: l.Name, path: l.Path,
leftover: true})
read = append(read, verdict{state: link.StateUnhealthy, reason: l.Why})
}
j.mu.Lock()
defer j.mu.Unlock()
changed := false
out := make([]link.ResourceHealth, 0, len(as))
for i, a := range as {
r := read[i]
// Every kept or waiting access lists its users, the tool runner's account with the uid read here (§4).
if r.entries == nil && len(a.users) > 0 && !a.leftover {
r.entries = entriesOf(a.users)
}
kind := link.KindAccess
if a.leftover {
kind = link.KindKeptEntries
}
v := link.ResourceHealth{Module: a.module, Resource: a.id, Kind: kind, Target: a.path,
State: r.state, Reason: r.reason, Since: now.UTC(), AccessWaits: r.waits, Act: r.act, Entries: r.entries}
if was, had := j.was[a.id]; had && was.State == r.state && was.Reason == r.reason {
v.Since = was.Since
v.Streak = was.Streak
} else {
changed = true
}
if r.state == link.StateUnhealthy {
v.Streak++
}
j.was[a.id] = v
out = append(out, v)
}
return out, changed
}
// withAccesses adds to a statement every access, as its module's resource of kind access.
func withAccesses(h *link.Health, vs []link.ResourceHealth) *link.Health {
h.Resources = append(h.Resources, vs...)
return h
}
// accessesHealthy says no access in vs is anything but healthy.
func accessesHealthy(vs []link.ResourceHealth) bool {
for _, v := range vs {
if v.State != link.StateHealthy {
return false
}
}
return true
}
// keeper runs the walks under kept folders beside the applies, one at a time (ADR 0325 §8): poked after every
// apply and every look, it walks whatever folder's entries under it are not what was declared.
type keeper struct {
running, following atomic.Bool
say func(string)
watch *apply.Watcher
}
// keeping is the one keeper on this machine.
var keeping = &keeper{say: func(string) {}}
// poke starts a reconcile of the kept folders unless one is running, and has the watch follow what is kept.
func (k *keeper) poke(ctx context.Context) {
// The watch is set beside, never in the caller's way: watching a new folder's tree reads every folder of it.
if k.following.CompareAndSwap(false, true) {
go func() {
defer k.following.Store(false)
k.follow()
}()
}
if !k.running.CompareAndSwap(false, true) {
return
}
go func() {
defer k.running.Store(false)
for _, line := range apply.ReconcileKept(ctx) {
k.say(line)
}
}()
}
// follow has the watch watch every kept folder with its users.
func (k *keeper) follow() {
if k.watch == nil {
return
}
all, err := apply.KeptFolders.All()
if err != nil {
return
}
folders := map[string][]store.KeptUser{}
for path, f := range all {
if f.Kept && len(f.Want) > 0 {
folders[path] = f.Want
}
}
k.watch.Set(folders)
}
// startKeeping opens the watch and has the keeper say what it does. A machine without the watch is said so, and
// its sweep alone finds what is made.
func startKeeping(ctx context.Context, say func(string)) {
keeping.say = say
w, err := apply.NewWatcher()
if err != nil {
say(fmt.Sprintf("the kept folders are not watched (%v): the sweep alone finds what is made in them", err))
} else {
keeping.watch = w
go w.Run(ctx)
}
keeping.poke(ctx)
}
// accessActFrom is the take act as the controller asks it over the link (ADR 0325 §8): judged against the
// declaration this node keeps, and run beside the applies.
func accessActFrom(statePath string, timeout time.Duration) link.AccessActFunc {
return func(act link.AccessAct) (string, error) {
kept, err := store.ReadDeclared(store.DeclaredPath(statePath))
if errors.Is(err, store.ErrNothingDeclared) {
return "", errors.New("this node keeps no declaration, so it is declared no access")
}
if err != nil {
return "", err
}
d, err := declaration.Parse(kept.Declaration)
if err != nil {
return "", fmt.Errorf("the declaration this node keeps does not read: %w", err)
}
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
_, said, err := apply.TakeFolder(ctx, d, act.Access, filepath.Clean(act.Path), act.Warrant, act.By,
apply.ExecRunner)
if err != nil {
fmt.Fprintf(os.Stderr, "mesh-host: the take act on %s was refused: %v\n", act.Access, err)
return "", err
}
return said, nil
}
}
// withWaiting is held with every id of waiting beside it: what liveness does not judge.
func withWaiting(held, waiting map[string]bool) map[string]bool {
if len(waiting) == 0 {
return held
}
out := map[string]bool{}
for id := range held {
out[id] = true
}
for id := range waiting {
out[id] = true
}
return out
}
-117
View File
@@ -1,117 +0,0 @@
package main
import (
"context"
"encoding/json"
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/novox/mesh-host/internal/apply"
"github.com/novox/mesh-host/internal/declaration"
"github.com/novox/mesh-host/internal/link"
"github.com/novox/mesh-host/internal/store"
)
// Row 4: the listing of a kept folder says, beside the tool runner's entry, that it stands for every module whose
// tools use the folder at the widest use (novox/hq ADR 0325 §4).
func TestTheListingSaysWhatTheToolRunnersEntryStandsFor(t *testing.T) {
entries := entriesOf([]store.KeptUser{
{Identity: "mesh-radarr", UID: 61001, Use: "read-write", From: declaration.FromContainer},
{Identity: "nox", UID: 61003, Use: "read-write", From: declaration.FromToolRunner},
})
if entries[0].Note != "" || !strings.Contains(entries[1].Note, "widest use") {
t.Fatalf("the listing: %+v", entries)
}
}
// Row 7, the engine's part: a folder that holds data and was never taken is said waiting with one wait of kind
// access — the access by its module's own id, the part its module, and words with no path (novox/hq ADR 0325 §7).
func TestAnAccessTheMeshMayNotKeepYetIsSaidAsAWaitOfItsOwnKind(t *testing.T) {
dir := t.TempDir()
lib := filepath.Join(dir, "movies")
if err := os.MkdirAll(filepath.Join(lib, "a film"), 0o755); err != nil {
t.Fatal(err)
}
// Run as root (the build seat's merge-check), the test's folders stand for an operator's, never root's.
if os.Geteuid() == 0 {
for _, p := range []string{dir, lib, filepath.Join(lib, "a film")} {
if err := os.Lchown(p, 65534, 65534); err != nil {
t.Fatal(err)
}
}
}
v := judgeAccess(judgedAccess{module: "radarr", id: "radarr.movies", name: "movies", path: lib,
users: []store.KeptUser{{Identity: "mesh-radarr", UID: 61001, Use: "read-write", From: "container"}}})
if v.state != link.StateWaiting || len(v.waits) != 1 {
t.Fatalf("not waiting with one wait: %+v", v)
}
raw, _ := json.Marshal(v.waits[0])
want := `{"kind":"access","access":"movies","part":"radarr","what":"let the mesh keep the shared folder movies",` +
`"placed":"` + v.waits[0].Placed + `","reason":"keep"}`
if string(raw) != want || !strings.HasPrefix(v.waits[0].Placed, "fs-") {
t.Fatalf("the wait reads %s", raw)
}
if strings.Contains(v.reason, dir) {
t.Fatalf("the reason names the path: %s", v.reason)
}
}
// An older controller's accesses (no users, no identities) are judged by nothing, as before ADR 0325: the statement
// carries no resource for them, waiting, unknown or kept-entries.
func TestAnOlderControllersAccessesSayNothing(t *testing.T) {
dir := t.TempDir()
d, err := declaration.Parse([]byte(`{"declaration":1,"resources":[
{"id":"radarr.library","type":"access","path":"` + filepath.Join(dir, "library") + `","mode":"read-write"},
{"id":"plex.movies","type":"access","path":"` + filepath.Join(dir, "absent") + `"}]}`))
if err != nil {
t.Fatal(err)
}
if err := os.Mkdir(filepath.Join(dir, "library"), 0o755); err != nil {
t.Fatal(err)
}
j := newAccessJudge()
j.set(d, nil)
if vs, _ := j.look(time.Now()); len(vs) != 0 {
t.Fatalf("an older controller's accesses are judged: %+v", vs)
}
}
// The waits' reasons and the tool runner's uid read here are said without words to parse (ADR 0325 §4, §7, §9).
func TestTheWaitsSayTheirReasonAndTheRunnersUid(t *testing.T) {
if link.ACLNeedsEnabling != apply.ACLNeedsEnabling || link.ACLUnsupported != apply.ACLUnsupported {
t.Fatal("the engine's ACL words are not the link's")
}
dir := t.TempDir()
was := apply.AccountIDs
apply.AccountIDs = func(context.Context, string) (int, int, bool, error) { return 61003, 1000, true, nil }
defer func() { apply.AccountIDs = was }()
d, err := declaration.Parse([]byte(`{"declaration":1,"resources":[{"id":"importer.books","type":"access",` +
`"path":"` + filepath.Join(dir, "no", "such") + `","mode":"read-write","users":[{"identity":"nox",` +
`"use":"read-write","from":"tool-runner"}]}]}`))
if err != nil {
t.Fatal(err)
}
j := newAccessJudge()
j.set(d, nil)
vs, _ := j.look(time.Now())
if len(vs) != 1 || len(vs[0].Entries) != 1 || vs[0].Entries[0].UID != 61003 {
t.Fatalf("the tool runner's uid read here is not said: %+v", vs)
}
// A folder whose disk is not there: its mount target named, not mounted.
disk := filepath.Join(dir, "disk")
if err := os.Mkdir(disk, 0o755); err != nil {
t.Fatal(err)
}
wasTargets := apply.MountTargets
apply.MountTargets = func(context.Context) []string { return []string{disk} }
defer func() { apply.MountTargets = wasTargets }()
v := judgeAccess(judgedAccess{module: "radarr", id: "radarr.movies", name: "movies",
path: filepath.Join(disk, "movies"), users: []store.KeptUser{{Identity: "mesh-radarr", UID: 61001,
Use: "read-write", From: "container"}}})
if len(v.waits) != 1 || v.waits[0].Reason != link.WaitDisk || link.WaitDisk != "disk" || link.WaitKeep != "keep" {
t.Fatalf("a disk wait does not say its reason: %+v", v.waits)
}
}
+274
View File
@@ -0,0 +1,274 @@
package main
import (
"context"
"errors"
"fmt"
"os"
"path/filepath"
"sort"
"sync"
"sync/atomic"
"time"
"github.com/novox/mesh-host/internal/apply"
"github.com/novox/mesh-host/internal/declaration"
"github.com/novox/mesh-host/internal/link"
"github.com/novox/mesh-host/internal/store"
)
// A library is judged on every look (novox/hq ADR 0328 §8, §9).
//
// Each library this machine is reached by is said as a resource `library:<name>` of kind library, with no module:
// healthy while it is kept (or, off its home, while its mount is there), waiting with an access wait of reason keep
// or disk, unknown while its filesystem could not be named, unhealthy otherwise. The root is the target, never in a
// reason nor in a wait's words: the controller says them on the operator's channel, which withholds a path (ADR
// 0234 §6). An access is judged by nothing, as before ADR 0325.
// libraryJudge reads every library on every look. Safe for the apply and the looking loop at once.
type libraryJudge struct {
mu sync.Mutex
libraries []declaration.Library
declared *declaration.Declaration
// was is each library's last verdict, with since when and how many looks in a row found it unhealthy.
was map[string]link.ResourceHealth
// reading guards inFlight: every library whose look has not returned yet.
reading sync.Mutex
inFlight map[string]bool
}
// accesses is the one library judge on this machine (its name from when it judged accesses).
var accesses = newLibraryJudge()
func newLibraryJudge() *libraryJudge {
return &libraryJudge{was: map[string]link.ResourceHealth{}, inFlight: map[string]bool{}}
}
// set is what the declaration just applied declares: every library. One no longer declared is forgotten.
func (j *libraryJudge) set(d *declaration.Declaration, _ map[string]bool) {
var ls []declaration.Library
if d != nil {
ls = append(ls, d.Libraries...)
}
sort.Slice(ls, func(a, b int) bool { return ls[a].Library < ls[b].Library })
j.mu.Lock()
defer j.mu.Unlock()
j.libraries, j.declared = ls, d
declared := map[string]bool{}
for _, l := range ls {
declared[apply.LibraryAccess(l.Library)] = true
}
for id := range j.was {
if !declared[id] {
delete(j.was, id)
}
}
}
// declaration is the declaration last set.
func (j *libraryJudge) declaration() *declaration.Declaration {
j.mu.Lock()
defer j.mu.Unlock()
return j.declared
}
// lookBound is how long a look waits for one library to answer. A hard network mount whose server is down blocks
// a stat for ever, and one library must never stop every health statement this machine makes.
var lookBound = 5 * time.Second
// verdict is one library's judgement in a look.
type verdict struct {
state, reason string
waits []link.AccessWait
act *link.AccessActRecord
}
// judgeBounded judges one library within lookBound, outside any lock. A look that has not answered is said so;
// while it still hangs, later looks start no other on the same library.
func (j *libraryJudge) judgeBounded(l declaration.Library) (verdict, bool) {
j.reading.Lock()
if j.inFlight[l.Library] {
j.reading.Unlock()
return verdict{}, false
}
j.inFlight[l.Library] = true
j.reading.Unlock()
done := make(chan verdict, 1)
go func() {
v := judgeLibrary(l)
j.reading.Lock()
delete(j.inFlight, l.Library)
j.reading.Unlock()
done <- v
}()
select {
case v := <-done:
return v, true
case <-time.After(lookBound):
return verdict{}, false
}
}
// judgeLibrary is one library's judgement now (ADR 0328 §7, §8, §9).
func judgeLibrary(l declaration.Library) verdict {
v := apply.JudgeLibrary(context.Background(), l)
access := apply.LibraryAccess(l.Library)
var out verdict
if a := v.Act; a != nil {
out.act = &link.AccessActRecord{Act: a.Act, Access: access, Warrant: a.Warrant, By: a.By, State: a.State,
Started: a.Started, Ended: a.Ended, Folders: a.Folders, Files: a.Files, Kept: a.Kept, Links: a.Links,
Other: a.Other, Mounts: a.Mounts, Shared: a.Shared, Why: a.Why}
}
switch v.Kind {
case apply.VerdictWaitKeep:
out.state, out.reason = link.StateWaiting, "its root holds data the mesh has not taken on the filesystem "+
"it is on now, and waits for the operator's word (novox/hq ADR 0328 §7, §8)"
out.waits = []link.AccessWait{{Kind: link.WaitKindAccess, Access: access, Part: l.Library,
What: apply.KeepWords(l.Library), Placed: v.Placed, Reason: link.WaitKeep}}
case apply.VerdictWaitDisk:
out.state, out.reason = link.StateWaiting, v.Why+" (novox/hq ADR 0328 §9)"
out.waits = []link.AccessWait{{Kind: link.WaitKindAccess, Access: access, Part: l.Library,
What: apply.DiskWords(l.Library), Reason: link.WaitDisk}}
case apply.VerdictUnknown:
// Not a fault and not the operator's: said as not known, and judged again on the next look.
out.state, out.reason = link.StateUnknown, v.Why+"; judged again on the next look"
case apply.VerdictBroken:
out.state, out.reason = link.StateUnhealthy, v.Why
default:
out.state = link.StateHealthy
switch {
case v.Act != nil && v.Act.State == link.ActRunning:
out.state, out.reason = link.StateStarting, "the take act is setting its group under its root"
case v.Act != nil && (v.Act.State == link.ActStopped || v.Act.State == link.ActFailed):
out.state, out.reason = link.StateStarting, "the take act goes on from where it stopped: "+v.Act.Why
}
}
return out
}
// look reads every library now. Whether any verdict changed since the last look, for the loop to say it.
func (j *libraryJudge) look(now time.Time) ([]link.ResourceHealth, bool) {
j.mu.Lock()
ls := append([]declaration.Library(nil), j.libraries...)
j.mu.Unlock()
read := make([]verdict, len(ls))
for i, l := range ls {
v, answered := j.judgeBounded(l)
if !answered {
v = verdict{state: link.StateUnknown, reason: "its root does not answer: judging it did not return " +
"within " + lookBound.String()}
}
read[i] = v
}
j.mu.Lock()
defer j.mu.Unlock()
changed := false
out := make([]link.ResourceHealth, 0, len(ls))
for i, l := range ls {
r := read[i]
id := apply.LibraryAccess(l.Library)
v := link.ResourceHealth{Resource: id, Kind: link.KindLibrary, Target: l.Root, State: r.state,
Reason: r.reason, Since: now.UTC(), AccessWaits: r.waits, Act: r.act}
if was, had := j.was[id]; had && was.State == r.state && was.Reason == r.reason {
v.Since = was.Since
v.Streak = was.Streak
} else {
changed = true
}
if r.state == link.StateUnhealthy {
v.Streak++
}
j.was[id] = v
out = append(out, v)
}
return out, changed
}
// withAccesses adds to a statement every library, as a resource of kind library.
func withAccesses(h *link.Health, vs []link.ResourceHealth) *link.Health {
h.Resources = append(h.Resources, vs...)
return h
}
// accessesHealthy says no library in vs is anything but healthy.
func accessesHealthy(vs []link.ResourceHealth) bool {
for _, v := range vs {
if v.State != link.StateHealthy {
return false
}
}
return true
}
// keeper goes on with take acts stopped at their bound, beside the applies, one at a time (ADR 0328 §7).
type keeper struct {
running atomic.Bool
say func(string)
}
// keeping is the one keeper on this machine.
var keeping = &keeper{say: func(string) {}}
// poke goes on with every stopped take act of the declaration last applied, unless that is running already.
func (k *keeper) poke(ctx context.Context) {
d := accesses.declaration()
if d == nil || len(d.Libraries) == 0 || !k.running.CompareAndSwap(false, true) {
return
}
go func() {
defer k.running.Store(false)
for _, line := range apply.ReconcileLibraries(ctx, d) {
k.say(line)
}
}()
}
// startKeeping has the keeper say what it does.
func startKeeping(ctx context.Context, say func(string)) {
keeping.say = say
keeping.poke(ctx)
}
// accessActFrom is the take act as the controller asks it over the link (ADR 0328 §7): judged against the
// declaration this node keeps, and run beside the applies.
func accessActFrom(statePath string, timeout time.Duration) link.AccessActFunc {
return func(act link.AccessAct) (string, error) {
kept, err := store.ReadDeclared(store.DeclaredPath(statePath))
if errors.Is(err, store.ErrNothingDeclared) {
return "", errors.New("this node keeps no declaration, so it is declared no library")
}
if err != nil {
return "", err
}
d, err := declaration.Parse(kept.Declaration)
if err != nil {
return "", fmt.Errorf("the declaration this node keeps does not read: %w", err)
}
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
_, said, err := apply.TakeLibrary(ctx, d, act.Access, filepath.Clean(act.Path), act.Warrant, act.By)
if err != nil {
fmt.Fprintf(os.Stderr, "mesh-host: the take act on %s was refused: %v\n", act.Access, err)
return "", err
}
return said, nil
}
}
// withWaiting is held with every id of waiting beside it: what liveness does not judge.
func withWaiting(held, waiting map[string]bool) map[string]bool {
if len(waiting) == 0 {
return held
}
out := map[string]bool{}
for id := range held {
out[id] = true
}
for id := range waiting {
out[id] = true
}
return out
}
+56
View File
@@ -0,0 +1,56 @@
package main
import (
"encoding/json"
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/novox/mesh-host/internal/apply"
"github.com/novox/mesh-host/internal/declaration"
"github.com/novox/mesh-host/internal/link"
)
// A library whose root holds data never taken is said waiting with one access wait: the access `library:<name>`, the
// part its name, words with no path, the filesystem's name, and reason keep (novox/hq ADR 0328 §8); one whose root
// is absent waits with reason disk (§9). Neither carries a module.
func TestALibrarySaysItsWaitWithoutWordsToParse(t *testing.T) {
was := apply.Libraries
apply.Libraries = apply.NewLibraryMemory()
defer func() { apply.Libraries = was }()
dir := t.TempDir()
root := filepath.Join(dir, "media")
if err := os.MkdirAll(filepath.Join(root, "a film"), 0o755); err != nil {
t.Fatal(err)
}
d, err := declaration.Parse([]byte(`{"declaration":1,"groups":[{"name":"mesh-media","gid":62000},` +
`{"name":"mesh-books","gid":62001}],"libraries":[` +
`{"library":"media","root":"` + root + `","group":"mesh-media","gid":62000,"home":true},` +
`{"library":"books","root":"` + filepath.Join(dir, "books") + `","group":"mesh-books","gid":62001,"home":true}],` +
`"owns_nothing":true,"resources":[]}`))
if err != nil {
t.Fatal(err)
}
j := newLibraryJudge()
j.set(d, nil)
vs, _ := j.look(time.Now())
if len(vs) != 2 {
t.Fatalf("the libraries are not said: %+v", vs)
}
books, media := vs[0], vs[1]
if media.Kind != link.KindLibrary || media.Resource != "library:media" || media.Module != "" ||
media.State != link.StateWaiting || len(media.AccessWaits) != 1 {
t.Fatalf("a library holding data is not said waiting: %+v", media)
}
raw, _ := json.Marshal(media.AccessWaits[0])
want := `{"kind":"access","access":"library:media","part":"media","what":"let the mesh keep the library media",` +
`"placed":"` + media.AccessWaits[0].Placed + `","reason":"keep"}`
if string(raw) != want || !strings.HasPrefix(media.AccessWaits[0].Placed, "fs-") {
t.Fatalf("the wait reads %s", raw)
}
if strings.Contains(media.Reason, dir) || len(books.AccessWaits) != 1 || books.AccessWaits[0].Reason != link.WaitDisk {
t.Fatalf("a missing root does not wait for its disk, or a reason names a path: %+v %+v", books, media)
}
}
+15 -16
View File
@@ -241,8 +241,8 @@ func run(ctx context.Context, command string, opts options) error {
// found (novox/hq issue 339).
handedOverFrom(opts.state)
foundState.Store(opts.state)
// And what it knows of the shared folders it keeps is beside its state (novox/hq ADR 0325 §5).
apply.KeptFolders = store.NewKeptFile(opts.state)
// And what it knows of the libraries it keeps is beside its state (novox/hq ADR 0328 §6).
apply.Libraries = store.NewLibraryFile(opts.state)
// And the guard asks the running container runtimes where they keep their data (novox/hq issue 339).
apply.AskRuntimes = apply.ExecRunner
switch command {
@@ -1123,7 +1123,7 @@ func runLink(ctx context.Context, opts options) error {
// this process's, a `reconcile` run by hand, the installer's — reads it (novox/hq issue 224).
sched.RecordWindowsIn(apply.WindowsIn(filepath.Dir(opts.state)))
go sched.Run(ctx)
// The watch on every kept folder and the walks under them, beside the applies (novox/hq ADR 0325 §6, §8).
// The take act's walks that stopped at their bound go on beside the applies (novox/hq ADR 0328 §7).
startKeeping(ctx, func(line string) { say(line) })
// The one liveness judge on this machine (novox/hq ADR 0240), made before anything applies: every
@@ -1236,7 +1236,7 @@ func runLink(ctx context.Context, opts options) error {
// A read-only judgement of the directories this node is declared, asked by the controller (novox/hq
// issue 358): which the walk that follows no link another account could make would refuse.
JudgeDirectories: judgeDirectoriesFrom(opts.state),
// The take act on an access's folder, on the operator's warrant (novox/hq ADR 0325 §8).
// The take act on a library, on the operator's warrant (novox/hq ADR 0328 §7).
AccessAct: accessActFrom(opts.state, opts.timeout),
Say: say,
Timeout: opts.timeout,
@@ -1602,8 +1602,7 @@ const (
// judgeWhatRuns looks at every long-running resource every liveness.LookEvery and says the statement on
// the bus when a state changed, again every minute while one is not healthy, and every five minutes
// anyway. A statement the bus did not take is said at the next look. It reads; it never acts (ADR 0240
// rule 6) — but for the sweep of a kept folder, which ADR 0325 §10 has every look make: it puts back an entry or a
// mask an app narrowed, and nothing else.
// rule 6) — but for going on with a take act stopped at its bound (ADR 0328 §7), which runs beside it.
func judgeWhatRuns(ctx context.Context, j *liveness.Judge, queue *link.Queue, say link.Announce) {
ticker := time.NewTicker(liveness.LookEvery)
defer ticker.Stop()
@@ -1642,8 +1641,8 @@ func judgeWhatRuns(ctx context.Context, j *liveness.Judge, queue *link.Queue, sa
}
}
}
// Every access, judged on every look, its kept folder swept (novox/hq ADR 0325 §10); and a walk stopped
// at its bound goes on.
// Every library, judged on every look (novox/hq ADR 0328 §8, §9); and a take act stopped at its bound
// goes on, beside.
accessSt, accessesChanged := accesses.look(time.Now())
keeping.poke(ctx)
owed = owed || accessesChanged
@@ -1700,10 +1699,10 @@ func judgeWhatRuns(ctx context.Context, j *liveness.Judge, queue *link.Queue, sa
// healthAsReported is a statement as the report and the event carry it.
func healthAsReported(st liveness.Statement, ns *network.Statement) *link.Health {
// IdentityContract: this engine reads a resource's declared `health` and judges it (ADR 0240 Phase B), judges a
// user's declared `root` (novox/hq ADR 0266), and reads a declaration's identities, a container's identity and
// an access's users (novox/hq ADR 0325), which is what tells the controller it may be sent any of those fields.
h := &link.Health{Contract: link.IdentityContract, At: st.At.UTC(), Resources: []link.ResourceHealth{}}
// LibraryContract: this engine reads a resource's declared `health` and judges it (ADR 0240 Phase B), judges a
// user's declared `root` (novox/hq ADR 0266), and reads a declaration's identities, groups and libraries
// (novox/hq ADR 0328), which is what tells the controller it may be sent any of those fields.
h := &link.Health{Contract: link.LibraryContract, At: st.At.UTC(), Resources: []link.ResourceHealth{}}
for _, r := range st.Resources {
h.Resources = append(h.Resources, link.ResourceHealth{Module: r.Module, Resource: r.ID, Kind: r.Kind,
Target: r.Target, State: r.State, Reason: r.Reason, Since: r.Since.UTC(), Streak: r.Streak,
@@ -2084,9 +2083,9 @@ func applyAndKeepHeld(ctx context.Context, opts options, raw []byte, signed *sto
// What runs here for a module, and how each is (novox/hq ADR 0240): judged from what was just
// applied, so every resource this apply started says `starting` and the gate waits for it.
if j := judging; j != nil {
// A container that waits for an accessed folder not there was not created: the access says why, and
// liveness judges it neither alive nor dead (novox/hq ADR 0325 §9).
j.Set(liveness.LongRunning(declared, withWaiting(held, apply.WaitingOnAccess(declared))))
// A container that waits for a library whose disk or mount is not there was not created: the library says
// why, and liveness judges it neither alive nor dead (novox/hq ADR 0328 §9).
j.Set(liveness.LongRunning(declared, withWaiting(held, apply.WaitingOnLibrary(declared))))
st, _ := j.Look(ctx)
netJudge.declare(declaredResolvConf(declared))
var netSt *network.Statement
@@ -2109,7 +2108,7 @@ func applyAndKeepHeld(ctx context.Context, opts options, raw []byte, signed *sto
accountSt = &as
}
accesses.set(declared, held)
// What the apply changed in who uses a kept folder is walked under it now, beside (ADR 0325 §8).
// A take act stopped at its bound goes on, beside (ADR 0328 §7).
keeping.poke(ctx)
accessSt, _ := accesses.look(time.Now())
report.Health = withAccesses(withFoundDirectories(withAccounts(withUnits(healthAsReported(st, netSt), unitSt),
+4 -4
View File
@@ -87,10 +87,10 @@ func TestTheStatementSaysAnAccountJudgedForRoot(t *testing.T) {
Groups: []string{"openrazer"}}, State: accounts.Healthy, Since: at},
}}
h := withAccounts(healthAsReported(liveness.Statement{At: at}, nil), as)
// At RootContract or later: an engine that also reads identities and keeps shared folders says IdentityContract
// (novox/hq ADR 0325), one above ADR 0312's GrantContract.
if h.Contract < link.RootContract || link.RootContract != 3 || h.Contract != link.IdentityContract ||
link.IdentityContract != link.GrantContract+1 {
// At RootContract or later: an engine that also reads libraries says LibraryContract (novox/hq ADR 0328), one
// above ADR 0325's IdentityContract.
if h.Contract < link.RootContract || link.RootContract != 3 || h.Contract != link.LibraryContract ||
link.LibraryContract != link.IdentityContract+1 {
t.Fatalf("contract %d", h.Contract)
}
got := map[string]string{}