Ask again for a folder that moved, and carry entries left while a disk was away (hq ADR 0325)

The engine names the filesystem a waiting folder is on; a warrant is recorded for that filesystem, and a
folder on another is asked about again and not taken there meanwhile. Entries the engine could not take
out while their disk was away, stated before a send, do not hold that delivery's gate. mesh-host pinned
to d2d9fbbd.
This commit is contained in:
2026-10-12 19:24:34 +02:00
parent 63602264ef
commit 2b15cb57f0
14 changed files with 195 additions and 56 deletions
+40 -29
View File
@@ -48,24 +48,27 @@ type folderTake struct {
// Access is the resource id of one access to it in the machine's declaration: what the take act names.
Access string `json:"access"`
// Accesses are the access ids (each module's own) that place it, Modules the modules waiting on it, both sorted.
Accesses []string `json:"accesses"`
Modules []string `json:"modules"`
At time.Time `json:"at"`
Accesses []string `json:"accesses"`
Modules []string `json:"modules"`
// Placed is the filesystem the folder is on, as the engine names it without a path: the warrant is for that one.
Placed string `json:"placed,omitempty"`
At time.Time `json:"at"`
}
// takeVerb is the verb a take's answer binds: the controller's own, performed by itself on the warrant.
const takeVerb = askerName + ".take"
// key is the folder on its machine: what one ask is kept for.
func (t folderTake) key() string { return t.Node + "\x00" + filepath.Clean(t.Path) }
// The filesystem is part of it: a folder that moved to another disk is asked about again.
func (t folderTake) key() string { return t.Node + "\x00" + filepath.Clean(t.Path) + "\x00" + t.Placed }
// digest binds everything the take does and was said for: the machine, the folder, the access it names and the
// modules the operator was told.
func (t folderTake) digest() string {
raw, _ := json.Marshal(struct {
Node, Path, Access string
Modules []string
}{t.Node, filepath.Clean(t.Path), t.Access, t.Modules})
Node, Path, Access, Placed string
Modules []string
}{t.Node, filepath.Clean(t.Path), t.Access, t.Placed, t.Modules})
sum := sha256.Sum256(raw)
return "sha256:" + hex.EncodeToString(sum[:])
}
@@ -246,7 +249,7 @@ func takeFolderIn(open *stores, ask func(ctx context.Context, act link.AccessAct
return "", fmt.Errorf("%s refused to take the folder: %s", t.Node, answer.Refused)
}
if err := open.inventory.RecordTaken(ctx, t.Node, inventory.TakenFolder{Path: t.Path, By: by, Ask: askID,
Warrant: askID, Said: answer.Said}); err != nil {
Warrant: askID, Said: answer.Said, Placed: t.Placed}); err != nil {
return "", fmt.Errorf("%s started the take (%s), and it could NOT be recorded: %w", t.Node, answer.Said, err)
}
return fmt.Sprintf("%s takes the folder for %s: %s", t.Node, strings.Join(t.Modules, ", "), answer.Said), nil
@@ -256,8 +259,9 @@ func takeFolderIn(open *stores, ask func(ctx context.Context, act link.AccessAct
// waitingFolder is one access wait excused for a folder not yet taken, as foldersWaiting gathers them.
type waitingFolder struct {
module, access string
// resource is the access's resource id on the machine, as the engine stated it.
resource string
// resource is the access's resource id on the machine, as the engine stated it; placed the filesystem its folder
// is on, without its path.
resource, placed string
}
// foldersWaiting is every folder waiting to be kept across the mesh (ADR 0325 §7, §8): from each machine's newest
@@ -302,7 +306,7 @@ func foldersWaiting(ctx context.Context, inv *inventory.Inventory, healths map[s
out[machine] = map[string][]waitingFolder{}
}
out[machine][path] = append(out[machine][path], waitingFolder{module: r.Module, access: w.Access,
resource: r.Resource})
resource: r.Resource, placed: w.Placed})
}
}
}
@@ -327,26 +331,33 @@ func foldersIn(open *stores) func(ctx context.Context) ([]folderTake, error) {
func takesOf(waiting map[string]map[string][]waitingFolder) []folderTake {
var out []folderTake
for machine, folders := range waiting {
for path, ws := range folders {
t := folderTake{Node: machine, Path: path}
seenM, seenA := map[string]bool{}, map[string]bool{}
var resources []string
for _, w := range ws {
if !seenM[w.module] {
seenM[w.module] = true
t.Modules = append(t.Modules, w.module)
for path, all := range folders {
// One take per filesystem the waits name: they agree but for a moment while a disk is swapped.
byPlaced := map[string][]waitingFolder{}
for _, w := range all {
byPlaced[w.placed] = append(byPlaced[w.placed], w)
}
for placed, ws := range byPlaced {
t := folderTake{Node: machine, Path: path, Placed: placed}
seenM, seenA := map[string]bool{}, map[string]bool{}
var resources []string
for _, w := range ws {
if !seenM[w.module] {
seenM[w.module] = true
t.Modules = append(t.Modules, w.module)
}
if !seenA[w.access] {
seenA[w.access] = true
t.Accesses = append(t.Accesses, w.access)
}
resources = append(resources, w.resource)
}
if !seenA[w.access] {
seenA[w.access] = true
t.Accesses = append(t.Accesses, w.access)
}
resources = append(resources, w.resource)
sort.Strings(t.Modules)
sort.Strings(t.Accesses)
sort.Strings(resources)
t.Access = resources[0]
out = append(out, t)
}
sort.Strings(t.Modules)
sort.Strings(t.Accesses)
sort.Strings(resources)
t.Access = resources[0]
out = append(out, t)
}
}
sort.Slice(out, func(i, j int) bool { return out[i].key() < out[j].key() })
+66 -3
View File
@@ -33,21 +33,28 @@ func keepWait(access string) inventory.Wait {
return inventory.Wait{Part: "radarr", Access: access, What: "let the mesh keep the shared folder " + access}
}
func placedWait(access, placed string) inventory.Wait {
w := keepWait(access)
w.Placed = placed
return w
}
func diskWait(access string) inventory.Wait {
return inventory.Wait{Part: "radarr", Access: access, Disk: true, What: "mount the disk that holds " + access}
}
// accessFacts is the facts with each folder given taken on the filesystem fs-000000000001.
func accessFacts(taken ...string) operatorWaitFacts {
kept := map[string]bool{}
kept := map[string]string{}
for _, p := range taken {
kept[p] = true
kept[p] = "fs-000000000001"
}
return operatorWaitFacts{
manifests: map[string]catalogue.Manifest{"radarr": radarrManifest, "plex": plexManifest},
given: map[string]map[string]time.Time{"radarr@ace": {}, "plex@ace": {}},
placed: map[string]map[string]string{"radarr@ace": {"movies": moviesAt}, "plex@ace": {"films": moviesAt},
"radarr@novox": {}},
taken: map[string]map[string]bool{"ace": kept},
taken: map[string]map[string]string{"ace": kept},
}
}
@@ -66,6 +73,10 @@ func TestAnAccessWaitIsExcusedOnlyForAFolderTheMeshMayNotKeepYet(t *testing.T) {
{"its disk not there", "ace", diskWait("movies"), accessFacts(), ""},
{"its disk not there, the folder taken before", "ace", diskWait("movies"), accessFacts(moviesAt), ""},
{"already taken", "ace", keepWait("movies"), accessFacts(moviesAt), "taken on ace already"},
{"already taken on the filesystem it is on", "ace", placedWait("movies", "fs-000000000001"),
accessFacts(moviesAt), "taken on ace already"},
{"taken on another filesystem: it moved, and is asked about again", "ace",
placedWait("movies", "fs-0000000000ff"), accessFacts(moviesAt), ""},
{"undeclared", "ace", keepWait("books"), accessFacts(), "does not declare"},
{"unplaced", "novox", keepWait("movies"), accessFacts(), "not placed on novox"},
{"taken folders unread", "ace", keepWait("movies"), operatorWaitFacts{manifests: accessFacts().manifests,
@@ -358,3 +369,55 @@ func TestTheDeclarationSaysItsIdentitiesOnlyWhenComposed(t *testing.T) {
t.Fatalf("the contract the engine states for ADR 0325 is 5, and this controller reads %d", link.IdentityContract)
}
}
// A folder that moved to another filesystem is asked about again: one ask per folder, machine and filesystem, and the
// take records the filesystem it was for.
func TestAFolderThatMovedIsAskedAboutAgain(t *testing.T) {
placed := "fs-000000000001"
r, taken := takeRig(t, func() map[string]map[string][]waitingFolder {
return waitingOn("ace", waitingFolder{module: "radarr", access: "movies", resource: "radarr.access-movies",
placed: placed})
})
ctx := context.Background()
if err := r.a.reconcile(ctx); err != nil {
t.Fatal(err)
}
placed = "fs-0000000000ff"
if err := r.a.reconcile(ctx); err != nil {
t.Fatal(err)
}
var open []asked
for _, a := range takeAsks(r) {
if a.State == askOpen {
open = append(open, a)
}
}
if len(open) != 1 || open[0].Take.Placed != "fs-0000000000ff" || len(takeAsks(r)) != 2 {
t.Fatalf("after the folder moved: %d asks, open %+v", len(takeAsks(r)), open)
}
answerWith(t, r, warrantOn(open[0], "let-the-mesh-keep-it", r.now))
if len(*taken) != 1 {
t.Fatalf("taken %v", *taken)
}
}
// The mesh's entries left while a disk was away (kind kept-entries), stated before the send, do not hold a delivery
// that did not bring them; stated since, they do — and they may name an access no longer declared.
func TestEntriesLeftWhileADiskWasAwayDoNotHoldADeliveryThatDidNotBringThem(t *testing.T) {
now := time.Now()
since := now.Add(-time.Minute)
left := inventory.ResourceHealth{Module: "radarr", Resource: "radarr.access-gone~entries", Kind: link.KindKeptEntries,
Target: moviesAt, State: link.StateUnhealthy, Reason: "the mesh's entries could not be taken out: its disk is away",
Since: since.Add(-time.Hour)}
f := gateFacts{now: now, waits: accessFacts(), groupsAdded: map[string]bool{"radarr": false},
health: map[string]inventory.NodeHealth{"ace": {Node: "ace", HeardAt: now,
Resources: []inventory.ResourceHealth{left}}}}
if h, why := moduleHealthWord("radarr", "ace", since, f); h != healthPerson || !strings.Contains(why, "access-gone~entries") {
t.Fatalf("entries left before the send read %v %q; want carried in a passing verdict", h, why)
}
left.Since = now
f.health["ace"] = inventory.NodeHealth{Node: "ace", HeardAt: now, Resources: []inventory.ResourceHealth{left}}
if h, _ := moduleHealthWord("radarr", "ace", since, f); h != healthNotYet {
t.Fatalf("entries left by this send read %v; want judged unhealthy", h)
}
}
+15 -3
View File
@@ -81,7 +81,7 @@ func stateHealth(ctx context.Context, inv *inventory.Inventory, k *conditions.Ke
// access by its module's id, and whether it waits for a disk rather than for the operator's word.
for _, w := range r.AccessWaits {
kept.Waits = append(kept.Waits, inventory.Wait{Part: w.Part, Access: w.Access, Disk: w.WaitsForDisk(),
What: w.What})
Placed: w.Placed, What: w.What})
}
resources = append(resources, kept)
}
@@ -650,7 +650,7 @@ func moduleHealthWord(module, machine string, since time.Time, f gateFacts) (hea
if waits && !f.groupsAdded[module] {
waits = false
}
var found, onWaiting []string
var found, onWaiting, leftover []string
var forOperator []inventory.Wait
for _, r := range resources {
if r.Module != module {
@@ -673,6 +673,14 @@ func moduleHealthWord(module, machine string, since time.Time, f gateFacts) (hea
found = append(found, r.Resource)
continue
}
// **The mesh's entries left on a folder whose disk was away, stated before this send, are a condition this
// build found** (novox/hq ADR 0325 §8): they may name an access no longer declared, they wait for the disk,
// and a delivery that did not bring them is not held for them — said in the verdict, as a directory used as
// found is. Stated since the send, the send brought them, and they are judged as unhealthy.
if r.Kind == link.KindKeptEntries && r.State == link.StateUnhealthy && r.Since.Before(since) {
leftover = append(leftover, r.Resource)
continue
}
switch r.State {
case link.StateHealthy:
case link.StateStarting:
@@ -696,7 +704,7 @@ func moduleHealthWord(module, machine string, since time.Time, f gateFacts) (hea
reasonAfter(r.Reason))
}
}
if waits || len(found) > 0 || len(forOperator) > 0 || len(onWaiting) > 0 {
if waits || len(found) > 0 || len(forOperator) > 0 || len(onWaiting) > 0 || len(leftover) > 0 {
said := onWaiting
if waits {
said = append(said, wait)
@@ -708,6 +716,10 @@ func moduleHealthWord(module, machine string, since time.Time, f gateFacts) (hea
said = append(said, fmt.Sprintf("on %s, %s uses %s as found and waits for the operator to hand it over "+
"(`nox node hand-over %s <directory>` on the control-node)", machine, module, strings.Join(found, ", "), machine))
}
if len(leftover) > 0 {
said = append(said, fmt.Sprintf("on %s, the mesh's entries %s could not be taken out while their disk was "+
"away, stated before this send: they are taken out when it is back", machine, strings.Join(leftover, ", ")))
}
return healthPerson, strings.Join(said, "; ")
}
return healthGood, ""
+11 -8
View File
@@ -45,10 +45,11 @@ type operatorWaitFacts struct {
manifests map[string]catalogue.Manifest
given map[string]map[string]time.Time
// placed is, per "<module>@<machine>", where each access of the module is on that machine, by id (novox/hq ADR
// 0325 §7); taken is, per machine, every folder the operator let the mesh keep there. A machine absent from
// taken could not be read, and no access wait of it is excused.
// 0325 §7); taken is, per machine, every folder the operator let the mesh keep there, with the filesystem it was
// taken on (empty when the engine named none). A machine absent from taken could not be read, and no access wait
// of it is excused.
placed map[string]map[string]string
taken map[string]map[string]bool
taken map[string]map[string]string
}
// checkWait is nil when a wait is excused, and otherwise why not, in words. Pure.
@@ -119,7 +120,9 @@ func checkAccessWait(module, machine string, m catalogue.Manifest, w inventory.W
return fmt.Errorf("says it waits for the access %s, and which folders are taken on %s could not be read",
w.Access, machine)
}
if taken[path] {
// A folder taken on another filesystem than the one it is on now (it moved to another disk) is not taken by that
// warrant: the wait is excused, and the operator is asked again (ADR 0325 §8).
if on, was := taken[path]; was && (w.Placed == "" || on == w.Placed) {
return fmt.Errorf("says it waits for the access %s, whose folder is taken on %s already: a module that gains "+
"an access to a taken folder is given its entries without a wait", w.Access, machine)
}
@@ -284,7 +287,7 @@ func readWaitFacts(ctx context.Context, inv *inventory.Inventory, machine string
f.placed = map[string]map[string]string{}
}
if f.taken == nil {
f.taken = map[string]map[string]bool{}
f.taken = map[string]map[string]string{}
}
if inv == nil {
return
@@ -330,9 +333,9 @@ func readWaitFacts(ctx context.Context, inv *inventory.Inventory, machine string
}
if _, read := f.taken[machine]; !read {
if taken, err := inv.KeptFolders(ctx, machine); err == nil {
kept := map[string]bool{}
for path := range taken {
kept[path] = true
kept := map[string]string{}
for path, t := range taken {
kept[path] = t.Placed
}
f.taken[machine] = kept
}
+15
View File
@@ -9,6 +9,7 @@ import (
"fmt"
"io"
"os"
"path/filepath"
"slices"
"sort"
"strings"
@@ -1066,6 +1067,20 @@ func folderFacts(ctx context.Context, inv *inventory.Inventory, node string, pla
for path := range taken {
kept[path] = true
}
// **A folder that moved to another filesystem is not taken there until its new warrant** (ADR 0325 §8): the
// machine's newest statement names, without a path, the filesystem each folder waiting for the operator is on now.
if stated, had, err := inv.HealthOf(ctx, node); err != nil {
return false, nil, nil, err
} else if had {
for _, r := range stated.Resources {
for _, w := range r.Waits {
path := filepath.Clean(r.Target)
if t, was := taken[path]; was && w.Access != "" && !w.Disk && w.Placed != "" && w.Placed != t.Placed {
kept[path] = false
}
}
}
}
return true, identities, kept, nil
}