Files
mesh-host/internal/apply/keeper_linux.go
T
jschoubben d2d9fbbd7d Hold what mounts a missing disk, and say entries the mesh could not take out (hq ADR 0325 re-review)
A container was still created over an unmounted mount point, so its writes landed on
the filesystem below; a folder moved to another disk was kept on the old warrant; and
entries left by a drop or a move that could not run lived only in a log line. Every
container mounting an access whose disk is away now waits, the warrant holds only for
the filesystem it was carried out on (the wait names the new one, path-free), entries
left on a filesystem are recorded per mount and taken out when it is back, and every
leftover is said as unhealthy until it is. A hung stat is shared per path, and a
widened mask takes from other entries only the bits it adds.
2026-10-12 19:03:02 +02:00

416 lines
15 KiB
Go

//go:build linux
package apply
import (
"context"
"errors"
"fmt"
"path/filepath"
"sync"
"time"
"golang.org/x/sys/unix"
"github.com/novox/mesh-host/internal/declaration"
"github.com/novox/mesh-host/internal/store"
)
// The walks under a kept folder, beside the apply (novox/hq ADR 0325 §6, §8, §10).
//
// The take act adds the users' entries to everything under a folder on the operator's warrant; ReconcileKept adds a
// joining user's and removes exactly a dropped user's under a folder already taken; the sweep looks at a bounded
// part of each kept folder on every look, going on where the last stopped, and puts back what an app narrowed.
// One walk at a time per folder: a sweep finds a folder being walked and waits for the next look.
// folderLocks is one lock per kept folder, held by every walk of it.
var folderLocks sync.Map
func folderLock(path string) *sync.Mutex {
l, _ := folderLocks.LoadOrStore(path, &sync.Mutex{})
return l.(*sync.Mutex)
}
// actRecord is a walk's counts as the record keeps them.
func actRecord(act, warrant, by string, started time.Time, w *aclWalk, err error) *store.KeptAct {
c := w.counts
rec := &store.KeptAct{Act: act, Warrant: warrant, By: by, State: "done", Started: started, Ended: time.Now().UTC(),
Folders: c.Folders, Files: c.Files, Kept: c.Kept, Links: c.Links, Other: c.Other, Mounts: c.Mounts}
switch {
case errors.Is(err, ErrWalkBound):
rec.State, rec.Why = "stopped", "it met its bound of entries or time; it goes on at the next apply"
case err != nil:
rec.State, rec.Why = "failed", err.Error()
case w.Failed != "":
rec.Why = fmt.Sprintf("%d entries could not be given their entries: %s", c.Unrepaired, w.Failed)
}
return rec
}
// ReconcileKept walks every kept folder whose entries under it are not what the last apply declared: a user that
// joined has its entries added to everything under it, a dropped user — or one whose use changed — has exactly its
// own taken out (ADR 0325 §8). Bounded by ActLimits; a walk stopped is gone on with at the next call. A folder
// nobody uses any more is forgotten once its entries are out. What it did, in words, for the engine to say.
func ReconcileKept(ctx context.Context) []string {
all, err := KeptFolders.All()
if err != nil {
return []string{"the kept folders could not be read: " + err.Error()}
}
var said []string
for path, f := range all {
if ctx.Err() != nil {
break
}
if len(f.Left) > 0 {
said = append(said, dropLeft(path, f)...)
if f, _ = keptRecord(path); f.Path == "" {
continue
}
}
if !f.Kept || sameUsers(f.Users, f.Want) {
continue
}
said = append(said, reconcileOne(ctx, path, f)...)
}
return said
}
// dropLeft takes out the entries the mesh left on a filesystem no longer at the folder, once that filesystem holds
// it again (§8): exactly those users' entries, nothing else. Left on the filesystem the folder is kept on now, they
// are that folder's own, and its walk takes out whichever are no longer wanted. Until then they stay recorded, and
// said (Leftovers).
func dropLeft(path string, f store.KeptFolder) []string {
lock := folderLock(path)
lock.Lock()
defer lock.Unlock()
holding := ""
if _, err := lstatWithin(path); err == nil {
holding = mountHolding(path, MountPoints())
}
var said []string
var still []store.KeptLeft
var mine []store.KeptUser
for _, l := range f.Left {
switch {
case f.Kept && l.Mount == f.Mount:
mine = append(mine, l.Users...)
case l.Mount != holding:
still = append(still, l)
default:
drop := map[uint32]bool{}
for _, u := range l.Users {
drop[u.UID] = true
}
w, err := walkKept(path, ActLimits, machineStats, nil, dropVisit(drop))
if err != nil || w.Failed != "" {
still = append(still, l)
said = append(said, fmt.Sprintf("the kept folder %s: the entries left on a filesystem that is back "+
"were not all taken out, tried again: %s", path, firstOf(pathlessOr(err), w.Failed)))
continue
}
said = append(said, fmt.Sprintf("the kept folder %s: the entries left on a filesystem that is back "+
"were taken out (%d folders and %d files)", path, w.counts.Folders, w.counts.Files))
}
}
_ = KeptFolders.Update(path, func(k *store.KeptFolder, had bool) bool {
k.Left = still
for _, u := range mine {
if _, has := usersOf(k.Users)[u.UID]; !has {
k.Users = append(k.Users, u)
}
}
return had && ((k.Kept && (len(k.Users) > 0 || len(k.Want) > 0)) || len(k.Left) > 0)
})
return said
}
func pathlessOr(err error) string {
if err == nil {
return ""
}
return pathless(err)
}
func firstOf(a, b string) string {
if a != "" {
return a
}
return b
}
func reconcileOne(ctx context.Context, path string, f store.KeptFolder) []string {
lock := folderLock(path)
lock.Lock()
defer lock.Unlock()
// **Judged as an apply judges it, before every walk** (§9, §12): a disk not there leaves its mount point —
// an empty folder of root's on the filesystem below — at the path, and a walk there would write onto root's
// folder, or "drop" entries from a folder that never had them and forget the ones on the data. Nothing is
// walked then, and the record stays for the walk that can reach the data.
if v := assessFolder(ctx, path, f, false); v.err != nil || v.kind != folderKeep || v.moved {
why := "the disk that holds it is not there"
switch {
case v.err != nil:
why = pathless(v.err)
case v.kind == folderRoot:
why = RootWords(v.why)
case v.moved || v.kind == folderWaitKeep:
why = "what is at it now is not what the mesh kept there"
}
_ = KeptFolders.Update(path, func(k *store.KeptFolder, had bool) bool {
k.Act = &store.KeptAct{Act: "walk", State: "waiting", Started: time.Now().UTC(),
Why: "not walked: " + why + "; walked once it is"}
return had
})
return []string{fmt.Sprintf("the kept folder %s: not walked, %s", path, why)}
}
have, want := usersOf(f.Users), usersOf(f.Want)
drop := map[uint32]bool{}
adds := false
for uid, rw := range have {
if w, ok := want[uid]; !ok || w != rw {
drop[uid] = true
}
}
for uid, rw := range want {
if h, ok := have[uid]; !ok || h != rw {
adds = true
}
}
act, visit := "add", keepVisit(want)
switch {
case len(drop) > 0 && adds:
visit = dropThenKeep(drop, want)
case len(drop) > 0:
act, visit = "drop", dropVisit(drop)
}
started := time.Now().UTC()
w, err := walkKept(path, ActLimits, machineStats, nil, visit)
rec := actRecord(act, "", "", started, w, err)
_ = KeptFolders.Update(path, func(k *store.KeptFolder, _ bool) bool {
k.Act = rec
// Done only when every entry was reached and changed: one that failed is walked again.
if err == nil && w.Failed == "" {
k.Users = f.Want
}
return k.Kept && (len(k.Want) > 0 || len(k.Users) > 0)
})
return []string{fmt.Sprintf("the kept folder %s: %s walk %s (%d folders and %d files changed, %d had "+
"them)%s", path, act, rec.State, rec.Folders, rec.Files, rec.Kept, orWhy(rec.Why))}
}
func orWhy(why string) string {
if why == "" {
return ""
}
return ": " + why
}
// TakeFolder is the take act (ADR 0325 §8): on the operator's warrant, the users' entries added to the folder of
// the access id and everything under it. It judges the ask against the declaration last applied and the folder as
// it is, sets the filesystem's ACL type where a ZFS dataset allows none yet, and starts the walk: what it answers
// at once, and a channel closed when the walk is through. Refused, nothing is changed.
func TakeFolder(ctx context.Context, d *declaration.Declaration, id, path, warrant, by string,
run Runner) (<-chan struct{}, string, error) {
if d == nil {
return nil, "", errors.New("this engine has applied no declaration, and takes no folder it was not told of")
}
var access *declaration.Access
for _, r := range d.Resources {
if a, ok := r.(*declaration.Access); ok && a.ID == id {
access = a
}
}
switch {
case access == nil:
return nil, "", fmt.Errorf("the access %s is not declared on this machine", id)
case filepath.Clean(access.Path) != path:
return nil, "", fmt.Errorf("the access %s is declared at another folder than the one asked", id)
}
folders := foldersOf(d.Resources, refusedPlaces(d.Resources))
want := folders[path]
users := want.list()
if len(users) == 0 {
return nil, "", errors.New("no identity uses the folder, so there is nothing to keep")
}
rec, _ := keptRecord(path)
v := assessFolder(ctx, path, rec, false)
switch {
case v.err != nil:
return nil, "", v.err
case v.kind == folderWaitDisk:
return nil, "", errors.New("the disk that holds the folder is not there")
case v.kind == folderRoot:
return nil, "", errors.New(RootWords(v.why))
case v.kind == folderMake:
return nil, "", errors.New("the folder is not there; the next apply makes it, and asks nobody")
}
lock := folderLock(path)
if !lock.TryLock() {
return nil, "", errors.New("a walk of the folder is running; asked again once it is through, the act goes on")
}
said := ""
_, err := keepTop(path, usersOf(users))
if errors.Is(err, ErrNoACL) {
allowed, aerr := allowACLs(ctx, v.mount, run)
if aerr != nil {
lock.Unlock()
return nil, "", aerr
}
said = allowed + "; "
_, err = keepTop(path, usersOf(users))
}
if err != nil {
lock.Unlock()
return nil, "", err
}
started := time.Now().UTC()
_ = KeptFolders.Update(path, func(f *store.KeptFolder, had bool) bool {
if v.moved {
*f = movedRecord(*f, v.mount)
}
if !f.Kept {
f.Kept, f.How, f.At, f.Users = true, store.KeptByWarrant, started, nil
}
f.Warrant, f.By, f.Mount, f.Want, f.IDs = warrant, by, v.mount, users, want.ids
// The warrant is for the filesystem it was carried out on (§8).
f.TakenOn = v.mount
f.Act = &store.KeptAct{Act: "take", Warrant: warrant, By: by, State: "running", Started: started}
return true
})
done := make(chan struct{})
go func() {
defer close(done)
defer lock.Unlock()
w, err := walkKept(path, ActLimits, machineStats, nil, keepVisit(usersOf(users)))
rec := actRecord("take", warrant, by, started, w, err)
_ = KeptFolders.Update(path, func(f *store.KeptFolder, _ bool) bool {
f.Act = rec
if err == nil {
f.Users = users
}
return true
})
}()
return done, said + "the take act started under the warrant " + warrant + ": the entries of " +
usersWords(users) + " are added to everything under the folder, following no link and entering no other " +
"filesystem", nil
}
// allowACLs lets the filesystem at mount carry ACLs, inside the take act only (ADR 0325 §8): a ZFS dataset's
// acltype set to posixacl, the dataset read from the machine's mounts. Any other filesystem is refused, naming why.
func allowACLs(ctx context.Context, mount string, run Runner) (string, error) {
source, fstype := mountSourceOf(mount)
if fstype != "zfs" || source == "" {
return "", fmt.Errorf("the filesystem that holds the folder (%s) allows no access entries, and only a ZFS "+
"dataset's can be let by the act; the operator mounts it with ACLs", orUnknown(fstype))
}
if _, err := run(ctx, "zfs", "set", "acltype=posixacl", source); err != nil {
return "", fmt.Errorf("the ZFS dataset that holds the folder could not be let carry access entries "+
"(zfs set acltype=posixacl): %w", err)
}
return "set acltype=posixacl on the ZFS dataset " + source, nil
}
func orUnknown(s string) string {
if s == "" {
return "of a type the machine does not say"
}
return s
}
// KeptLook is what one look found of a kept folder (ADR 0325 §10).
type KeptLook struct {
// Short is every user the folder itself gives less than its right, by entry and mask together, in words.
Short []string
// Busy is a walk of the folder running: the sweep waited.
Busy bool
// Sweep is the sweep's place after this look.
Sweep store.KeptSweep
// Repaired is how many entries this look found narrowed and put back; Unrepaired how many it could not.
Repaired, Unrepaired int
// Failed is why the sweep could not look, in words with no path.
Failed string
}
// LookKept looks at a kept folder once: the folder itself judged for each user, then the sweep's bounded part of
// what is under it, from where the last look stopped, putting back what an app narrowed. The sweep's place is
// recorded; a look that ran out of its bound says so.
func LookKept(path string, users []store.KeptUser, limits WalkLimits) KeptLook {
var out KeptLook
short, err := judgeTop(path, users)
if err != nil {
out.Failed = err.Error()
return out
}
out.Short = short
lock := folderLock(path)
if !lock.TryLock() {
out.Busy = true
rec, _ := keptRecord(path)
out.Sweep = rec.Sweep
return out
}
defer lock.Unlock()
rec, _ := keptRecord(path)
sw := rec.Sweep
w, err := walkKept(path, limits, machineStats, sw.Cursor, keepVisit(usersOf(users)))
out.Repaired, out.Unrepaired = w.counts.Narrowed-w.counts.Unrepaired, w.counts.Unrepaired
sw.Read += w.counts.Read
sw.Narrowed += w.counts.Narrowed
sw.Unrepaired += w.counts.Unrepaired
switch {
case errors.Is(err, ErrWalkBound):
sw.Cursor, sw.Bound = w.Cursor, true
case err != nil:
out.Failed = err.Error()
default:
// Covered whole: the next look starts a new pass, and what this one could not put back is kept until
// the next pass is through.
sw = store.KeptSweep{Covered: time.Now().UTC(), LastUnrepaired: sw.Unrepaired}
}
out.Sweep = sw
_ = KeptFolders.Update(path, func(f *store.KeptFolder, had bool) bool {
f.Sweep = sw
return had
})
return out
}
// judgeTop is every user the folder itself gives less than its right, read only.
func judgeTop(path string, users []store.KeptUser) ([]string, error) {
var short []string
_, fd, st, err := openKept(path, WalkLimits{Entries: 1, Time: time.Minute}, machineStats, nil)
if err != nil {
return nil, err
}
defer unix.Close(fd)
a, err := getACL(fd, aclAccessXattr, uint32(st.Mode)&0o7777)
if err != nil {
return nil, err
}
for _, u := range users {
need := needOn(u.Use == declaration.AccessReadWrite, true)
if got := effective(a, u.UID, st.Uid, uint32(st.Mode)&0o7777); got&need != need {
short = append(short, fmt.Sprintf("%s has %s of the %s it needs", u.Identity, permWords(got),
permWords(need)))
}
}
return short, nil
}
// JudgeKept reads, without changing anything, how many entries under a kept folder give a user less than its
// right, within limits: what the access says while an entry an app narrowed is not put back yet (ADR 0325 §6).
func JudgeKept(path string, users []store.KeptUser, limits WalkLimits) (WalkCounts, error) {
um := usersOf(users)
w, err := walkKept(path, limits, machineStats, nil, func(fd int, st unix.Stat_t, folder bool) (bool, bool, bool,
error) {
a, err := getACL(fd, aclAccessXattr, uint32(st.Mode)&0o7777)
if err != nil {
return false, false, false, err
}
n, u := judgeACL(a, um, st, folder)
return false, n, u, nil
})
return w.counts, err
}