Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
77b48a4f1a | ||
|
|
e84a121b8a | ||
|
|
06b13ec1cc | ||
|
|
0830fcb25e | ||
|
|
b03bf4c118 | ||
|
|
53f2649369 | ||
|
|
9e5666b4d6 | ||
|
|
c006aa8242 |
+17
-14
@@ -22,20 +22,7 @@ func NodeCapacity(path, node string, logf func(string, ...any)) func() int {
|
||||
}
|
||||
}
|
||||
return func() int {
|
||||
if path == "" {
|
||||
return 0
|
||||
}
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
say("the build capacity cannot be read, so %s pulls nothing: %v", node, err)
|
||||
return 0
|
||||
}
|
||||
var s map[string]json.RawMessage
|
||||
if err := json.Unmarshal(raw, &s); err != nil {
|
||||
say("the settings file %s is not JSON, so %s pulls nothing: %v", path, node, err)
|
||||
return 0
|
||||
}
|
||||
n, err := capacityOf(s[CapacitySetting])
|
||||
n, err := ReadCapacity(path)
|
||||
if err != nil {
|
||||
say("%v, so %s pulls nothing", err, node)
|
||||
return 0
|
||||
@@ -44,6 +31,22 @@ func NodeCapacity(path, node string, logf func(string, ...any)) func() int {
|
||||
}
|
||||
}
|
||||
|
||||
// ReadCapacity is the capacity the settings file at path holds, and why it cannot be read when it cannot (then 0).
|
||||
func ReadCapacity(path string) (int, error) {
|
||||
if path == "" {
|
||||
return 0, nil
|
||||
}
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("the build capacity cannot be read: %w", err)
|
||||
}
|
||||
var s map[string]json.RawMessage
|
||||
if err := json.Unmarshal(raw, &s); err != nil {
|
||||
return 0, fmt.Errorf("the settings file %s is not JSON: %w", path, err)
|
||||
}
|
||||
return capacityOf(s[CapacitySetting])
|
||||
}
|
||||
|
||||
// capacityOf is the capacity a config value holds: absent or null is 0; anything but a whole number of at least 0 is
|
||||
// refused, naming it.
|
||||
func capacityOf(value json.RawMessage) (int, error) {
|
||||
|
||||
+82
-8
@@ -98,6 +98,12 @@ type Puller struct {
|
||||
// again, and the worker it came from is not looked at again until the next pull's wait, so a node lacking the need
|
||||
// does not take the asker's next copy at once. Nil has nothing.
|
||||
Has func(need string) bool
|
||||
// SettingsFile is the module's settings file. With it set the puller wakes whenever the file is written (a file event,
|
||||
// WatchSettings), and reads Capacity from it itself, keeping why it cannot be read (CapacityNote); a parked puller at
|
||||
// capacity 0 therefore wakes when the node-engine refills the file, and a failed config follow is tried again then.
|
||||
SettingsFile string
|
||||
// watching, when set, is called each time the settings file's watch is (re)armed: a test waits on it, not on a sleep.
|
||||
watching func()
|
||||
|
||||
mu sync.Mutex
|
||||
inHand int
|
||||
@@ -108,6 +114,14 @@ type Puller struct {
|
||||
// watched is the capacity the config watch handed last, used in place of Capacity once hasWatched.
|
||||
watched int
|
||||
hasWatched bool
|
||||
// capFollow and hoursFollow are the state of the two config follows: followed once WatchConfig took them, trying
|
||||
// while a start is in flight; followFault is the last failure said, so each failure is said once.
|
||||
capFollowed, capTrying, hoursFollowed, hoursTrying bool
|
||||
followFault string
|
||||
// fileFault is why the settings file's capacity cannot be read, "" when it can.
|
||||
fileFault string
|
||||
// capWhy is why the config watch's last value could not be read, "" when it could.
|
||||
capWhy string
|
||||
// hours is the build window the config watch handed last, and hoursFault why its value cannot be read; used in place
|
||||
// of BuildHours once hasHours.
|
||||
hours *BuildWindow
|
||||
@@ -248,6 +262,21 @@ func (p *Puller) roomNow() (ok bool, capacity, inHand int) {
|
||||
return !p.paused && p.inHand < capacity, capacity, p.inHand
|
||||
}
|
||||
|
||||
// CapacityNote says when the puller takes nothing for want of a capacity: why, and whether it is a fault (a setting that
|
||||
// cannot be read or is refused, so a person must look) and not a choice (a capacity set to 0). "" when it pulls.
|
||||
func (p *Puller) CapacityNote() (why string, fault bool) {
|
||||
c := p.CapacityDetail()
|
||||
if c.Effective > 0 {
|
||||
return "", false
|
||||
}
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
if p.fileFault != "" && !p.hasWatched {
|
||||
return "the " + p.Language + " build capacity of " + p.Node + " is 0 because it cannot be read: " + p.fileFault, true
|
||||
}
|
||||
return "the " + p.Language + " build capacity of " + p.Node + " is " + c.Why, p.hasWatched && p.capWhy != ""
|
||||
}
|
||||
|
||||
// CapacityNow is the capacity the puller judges by (CapacityDetail's Effective).
|
||||
func (p *Puller) CapacityNow() int {
|
||||
return p.CapacityDetail().Effective
|
||||
@@ -265,6 +294,15 @@ func (p *Puller) CapacityDetail() delivery.Capacity {
|
||||
switch {
|
||||
case has:
|
||||
configured, from = watched, "mesh-config, as the config watch handed it"
|
||||
case p.SettingsFile != "":
|
||||
n, err := ReadCapacity(p.SettingsFile)
|
||||
configured = n
|
||||
p.mu.Lock()
|
||||
p.fileFault = ""
|
||||
if err != nil {
|
||||
p.fileFault = err.Error()
|
||||
}
|
||||
p.mu.Unlock()
|
||||
case p.Capacity != nil:
|
||||
configured = p.Capacity()
|
||||
}
|
||||
@@ -386,8 +424,8 @@ func (p *Puller) Pull(ctx context.Context) error {
|
||||
// node kept from building is held by its capacity (0), not by a pause.
|
||||
p.logf("the pause of %s cannot be read: %v; pulling as not paused", p.Node, err)
|
||||
}
|
||||
go p.watchCapacity() // WatchConfigServed waits for Serve, which a module starts after its pull
|
||||
go p.watchHours()
|
||||
p.follow() // WatchConfigServed waits for Serve, which a module starts after its pull
|
||||
go p.watchSettings(ctx)
|
||||
wait := p.PullWait
|
||||
if wait <= 0 {
|
||||
wait = DefaultPullWait
|
||||
@@ -414,6 +452,7 @@ func (p *Puller) Pull(ctx context.Context) error {
|
||||
case <-edge:
|
||||
}
|
||||
stop()
|
||||
p.follow() // a follow that failed is tried again at the next event
|
||||
_, capacity, inHand = p.roomNow()
|
||||
p.logf("the %s puller on %s wakes: %d of capacity %d in hand, paused %v", p.Language, p.Node, inHand, capacity, p.Paused())
|
||||
continue
|
||||
@@ -432,6 +471,7 @@ func (p *Puller) Pull(ctx context.Context) error {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-p.wake:
|
||||
p.follow()
|
||||
case <-time.After(wait):
|
||||
}
|
||||
continue
|
||||
@@ -620,11 +660,18 @@ func (p *Puller) watchCapacity() {
|
||||
if p.WatchConfig == nil {
|
||||
return
|
||||
}
|
||||
p.mu.Lock()
|
||||
if p.capFollowed || p.capTrying {
|
||||
p.mu.Unlock()
|
||||
return
|
||||
}
|
||||
p.capTrying = true
|
||||
p.mu.Unlock()
|
||||
err := p.WatchConfig(CapacitySetting, func(c stdio.ProvisionChange) error {
|
||||
n, why := p.capacityFrom(c)
|
||||
p.mu.Lock()
|
||||
before, had := p.watched, p.hasWatched
|
||||
p.watched, p.hasWatched = n, true
|
||||
p.watched, p.hasWatched, p.capWhy = n, true, why
|
||||
p.mu.Unlock()
|
||||
switch {
|
||||
case why != "":
|
||||
@@ -635,9 +682,31 @@ func (p *Puller) watchCapacity() {
|
||||
p.Wake()
|
||||
return nil
|
||||
})
|
||||
p.followed(&p.capFollowed, &p.capTrying, err, "build capacity")
|
||||
}
|
||||
|
||||
// follow starts each config follow that is not running: at the start, and at every wake after (a file event, a resume, a
|
||||
// step's end), so a follow that failed is tried again by the next event and not given up for good.
|
||||
func (p *Puller) follow() {
|
||||
go p.watchCapacity()
|
||||
go p.watchHours()
|
||||
}
|
||||
|
||||
// followed records the end of one follow's start. A failure is said once per distinct error, with that it is tried again.
|
||||
func (p *Puller) followed(done, trying *bool, err error, what string) {
|
||||
p.mu.Lock()
|
||||
*trying = false
|
||||
*done = err == nil
|
||||
said := err != nil && err.Error() == p.followFault
|
||||
if err != nil {
|
||||
p.logf("the %s build capacity of %s cannot be followed on the bus, so it is read from the settings file when the "+
|
||||
"puller wakes: %v", p.Language, p.Node, err)
|
||||
p.followFault = err.Error()
|
||||
} else {
|
||||
p.followFault = ""
|
||||
}
|
||||
p.mu.Unlock()
|
||||
if err != nil && !said {
|
||||
p.logf("the %s %s of %s cannot be followed on the bus, so it is read from the settings file; the follow is tried "+
|
||||
"again at the next event: %v", p.Language, what, p.Node, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -648,6 +717,13 @@ func (p *Puller) watchHours() {
|
||||
if p.WatchConfig == nil || p.BuildHours == nil {
|
||||
return
|
||||
}
|
||||
p.mu.Lock()
|
||||
if p.hoursFollowed || p.hoursTrying {
|
||||
p.mu.Unlock()
|
||||
return
|
||||
}
|
||||
p.hoursTrying = true
|
||||
p.mu.Unlock()
|
||||
err := p.WatchConfig(BuildHoursSetting, func(c stdio.ProvisionChange) error {
|
||||
setting := ""
|
||||
var w *BuildWindow
|
||||
@@ -672,9 +748,7 @@ func (p *Puller) watchHours() {
|
||||
p.Wake()
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
p.logf("the %s build hours of %s cannot be followed on the bus, so they are read from the settings file: %v", p.Language, p.Node, err)
|
||||
}
|
||||
p.followed(&p.hoursFollowed, &p.hoursTrying, err, "build hours")
|
||||
}
|
||||
|
||||
// WatchConfigServed is stdio.WatchConfig once Serve has the runtime's channel (stdio.Served): a build module starts its
|
||||
|
||||
@@ -5,6 +5,8 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@@ -301,6 +303,71 @@ func TestPullAtCapacityZeroTakesNothing(t *testing.T) {
|
||||
w.set(t, "2")
|
||||
waitFor(t, "the ask taken once the capacity moved to 2", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
||||
})
|
||||
t.Run("settings refill wakes a parked puller and retries a failed follow", func(t *testing.T) {
|
||||
// 2026-10-28: a replaced builder held the old "shanks=2,..." in its settings file, its config follow timed out, and
|
||||
// the puller parked at 0 with nothing to wake it when the node-engine refilled the file.
|
||||
src, st := NewFakeAsks(), newSteps()
|
||||
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
||||
file := filepath.Join(t.TempDir(), "settings.json")
|
||||
if err := os.WriteFile(file, []byte(`{"capacity":"shanks=2,ace=2"}`), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var mu sync.Mutex
|
||||
attempts := 0
|
||||
watch := func(key string, handle func(stdio.ProvisionChange) error) error {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
attempts++
|
||||
if attempts == 1 {
|
||||
return errors.New("nats: timeout")
|
||||
}
|
||||
return nil // followed; no value handed yet
|
||||
}
|
||||
armed := make(chan struct{}, 8)
|
||||
var parked atomic.Int32
|
||||
parks := func() int32 { return parked.Load() }
|
||||
logf := func(format string, args ...any) {
|
||||
if strings.Contains(format, "parks") {
|
||||
parked.Add(1)
|
||||
}
|
||||
}
|
||||
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, PullWait: 10 * time.Millisecond,
|
||||
SettingsFile: file, WatchConfig: watch, Logf: logf, watching: func() { armed <- struct{}{} },
|
||||
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
go p.Pull(ctx)
|
||||
waitFor(t, "the parked puller to say why", func() bool { why, fault := p.CapacityNote(); return fault && why != "" })
|
||||
if src.Waiting(delivery.AskDelivery) != 1 {
|
||||
t.Fatal("a node whose capacity cannot be read took an ask")
|
||||
}
|
||||
<-armed // the file watch is in place
|
||||
tmp := file + ".new"
|
||||
if err := os.WriteFile(tmp, []byte(`{"capacity":2}`), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Rename(tmp, file); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitFor(t, "the ask taken once the file was refilled", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
||||
// and the directory itself replaced (IN_IGNORED): the watch is armed again and a later refill still wakes the puller
|
||||
dir := filepath.Dir(file)
|
||||
before := parks()
|
||||
if err := os.RemoveAll(dir); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Mkdir(dir, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
<-armed // armed again
|
||||
waitFor(t, "the puller parked with no settings file", func() bool { return parks() > before })
|
||||
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
||||
if err := os.WriteFile(file, []byte(`{"capacity":2}`), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitFor(t, "the second ask taken after the directory was replaced", func() bool { a, _, _ := src.Fates(); return len(a) == 2 })
|
||||
waitFor(t, "the failed follow tried again", func() bool { mu.Lock(); defer mu.Unlock(); return attempts >= 2 })
|
||||
})
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
watch func(string, func(stdio.ProvisionChange) error) error
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
//go:build linux
|
||||
|
||||
package buildstep
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
const settingsMask = syscall.IN_CLOSE_WRITE | syscall.IN_MOVED_TO | syscall.IN_CREATE
|
||||
|
||||
// watchSettings wakes the puller each time its settings file is written or replaced (inotify on the file's directory, so
|
||||
// a rename over it counts): the event the node-engine's refill of the file is, which a parked puller at capacity 0
|
||||
// otherwise never sees. No timer: it blocks on the kernel's event. When the kernel drops the watch (IN_IGNORED: the
|
||||
// directory was replaced or removed) it is added again, said once; while the directory is missing the nearest existing
|
||||
// ancestor is watched for the directory's return. A watch that cannot start is said once.
|
||||
func (p *Puller) watchSettings(ctx context.Context) {
|
||||
if p.SettingsFile == "" {
|
||||
return
|
||||
}
|
||||
fd, err := syscall.InotifyInit1(syscall.IN_CLOEXEC | syscall.IN_NONBLOCK)
|
||||
if err != nil {
|
||||
p.logf("the settings file %s cannot be watched, so a refill of it wakes the puller only at another event: %v", p.SettingsFile, err)
|
||||
return
|
||||
}
|
||||
f := os.NewFile(uintptr(fd), "inotify")
|
||||
defer f.Close()
|
||||
dir, base := filepath.Split(p.SettingsFile)
|
||||
dir = filepath.Clean(dir)
|
||||
// arm watches dir, or when it is missing its nearest existing ancestor; it returns the watch descriptor and whether it
|
||||
// is the directory's own.
|
||||
arm := func() (int, bool, error) {
|
||||
wd, err := syscall.InotifyAddWatch(fd, dir, settingsMask)
|
||||
if err == nil {
|
||||
return wd, true, nil
|
||||
}
|
||||
for up := filepath.Dir(dir); ; up = filepath.Dir(up) {
|
||||
wd, aerr := syscall.InotifyAddWatch(fd, up, settingsMask)
|
||||
if aerr == nil {
|
||||
// the directory may have come back between the two calls: look once more, now that the ancestor is watched
|
||||
if own, oerr := syscall.InotifyAddWatch(fd, dir, settingsMask); oerr == nil {
|
||||
return own, true, nil
|
||||
}
|
||||
return wd, false, nil
|
||||
}
|
||||
if up == filepath.Dir(up) {
|
||||
return 0, false, err
|
||||
}
|
||||
}
|
||||
}
|
||||
wd, own, err := arm()
|
||||
if err != nil {
|
||||
p.logf("the settings file %s cannot be watched, so a refill of it wakes the puller only at another event: %v", p.SettingsFile, err)
|
||||
return
|
||||
}
|
||||
if p.watching != nil {
|
||||
p.watching()
|
||||
}
|
||||
go func() { <-ctx.Done(); f.Close() }()
|
||||
buf := make([]byte, 4096)
|
||||
for {
|
||||
n, err := f.Read(buf)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
hit, rearm := false, false
|
||||
for off := 0; off+16 <= n; {
|
||||
ewd := int(int32(binary.LittleEndian.Uint32(buf[off:])))
|
||||
mask := binary.LittleEndian.Uint32(buf[off+4:])
|
||||
l := int(binary.LittleEndian.Uint32(buf[off+12:]))
|
||||
name := buf[off+16 : off+16+l]
|
||||
for i, c := range name {
|
||||
if c == 0 {
|
||||
name = name[:i]
|
||||
break
|
||||
}
|
||||
}
|
||||
switch {
|
||||
case mask&syscall.IN_IGNORED != 0 && ewd == wd:
|
||||
rearm = true
|
||||
case ewd == wd && own && string(name) == base:
|
||||
hit = true
|
||||
case ewd == wd && !own:
|
||||
rearm = true // something appeared where the directory was: look again
|
||||
}
|
||||
off += 16 + l
|
||||
}
|
||||
if rearm {
|
||||
nwd, nown, err := arm()
|
||||
if err != nil {
|
||||
p.logf("the settings file %s is no longer watched, so a refill of it wakes the puller only at another event: %v", p.SettingsFile, err)
|
||||
return
|
||||
}
|
||||
if own || nown { // said once per loss of the watch on the directory itself
|
||||
if own {
|
||||
p.logf("the watch of the settings file %s was dropped (its directory was replaced); watching again", p.SettingsFile)
|
||||
}
|
||||
}
|
||||
wd, own = nwd, nown
|
||||
if p.watching != nil {
|
||||
p.watching()
|
||||
}
|
||||
hit = true // the file may have been refilled meanwhile
|
||||
}
|
||||
if hit {
|
||||
p.Wake()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
//go:build !linux
|
||||
|
||||
package buildstep
|
||||
|
||||
import "context"
|
||||
|
||||
// watchSettings watches nothing off Linux: the puller wakes at its other events.
|
||||
func (p *Puller) watchSettings(ctx context.Context) {}
|
||||
@@ -99,6 +99,7 @@ func Answer(ctx context.Context, r Reader, mesh Mesh, s Scope, fps map[string]st
|
||||
for _, m := range sortedKeys(failed) {
|
||||
a.Refusals = append(a.Refusals, delivery.ModuleRefused{Module: m, Why: failed[m]})
|
||||
}
|
||||
heldAsToday(&a, s.HeldToday)
|
||||
a.Verdict = delivery.VerdictOf(a)
|
||||
return a
|
||||
}
|
||||
@@ -203,10 +204,31 @@ func Answer(ctx context.Context, r Reader, mesh Mesh, s Scope, fps map[string]st
|
||||
a.Refusals = append(a.Refusals, delivery.ModuleRefused{Module: m, Why: failed[m]})
|
||||
}
|
||||
sort.Slice(a.Refusals, func(i, j int) bool { return a.Refusals[i].Module < a.Refusals[j].Module })
|
||||
heldAsToday(&a, s.HeldToday)
|
||||
a.Verdict = delivery.VerdictOf(a)
|
||||
return a
|
||||
}
|
||||
|
||||
// heldAsToday moves each refusal of a module the node holds today (Scope.HeldToday) into Held: the candidate holds it as
|
||||
// the node does, which is no change (issue 1126: a seed reproduces today, and today holds it too).
|
||||
func heldAsToday(a *delivery.CandidateAnswer, today []string) {
|
||||
if len(today) == 0 {
|
||||
return
|
||||
}
|
||||
kept := a.Refusals[:0]
|
||||
for _, r := range a.Refusals {
|
||||
if slices.Contains(today, r.Module) {
|
||||
a.Held = append(a.Held, r)
|
||||
} else {
|
||||
kept = append(kept, r)
|
||||
}
|
||||
}
|
||||
a.Refusals = kept
|
||||
if len(a.Refusals) == 0 {
|
||||
a.Refusals = nil
|
||||
}
|
||||
}
|
||||
|
||||
// maxCulprits bounds the modules one refusal is tried on, each a whole composition.
|
||||
const maxCulprits = 8
|
||||
|
||||
|
||||
@@ -91,6 +91,19 @@ func answerNamesEveryRefusalAndLeavesAnOldBuildToTheNode(t *testing.T) {
|
||||
if lemursOK := slices.ContainsFunc(a.Added, func(m delivery.ModuleAdded) bool { return m.Module == "lemurs" }); !lemursOK {
|
||||
t.Errorf("lemurs, which composes, is not answered beside the refusals: %+v", a)
|
||||
}
|
||||
// Issue 1126: a module the node holds today that the candidate holds too is held as today, no refusal: with every
|
||||
// refused module held today the verdict is no refusal, and each is named under Held.
|
||||
today := Answer(ctx, b, Mesh{}, Scope{Node: "n1", Placements: given(b), HeldToday: []string{"api", "spiegel", "web"}},
|
||||
map[string]string{"lemurs": fp("e"), "web": fp("1"), "api": fp("2"), "spiegel": fp("3")}, nil, "p-h1", 3)
|
||||
if today.Verdict == delivery.CandidateRefused || len(today.Refusals) != 0 || len(today.Held) != 3 || today.Held[0].Why == "" {
|
||||
t.Errorf("held as today: %+v", today)
|
||||
}
|
||||
// A refused module the node does not hold today stays a refusal: holding only api today leaves spiegel and web refused.
|
||||
part := Answer(ctx, b, Mesh{}, Scope{Node: "n1", Placements: given(b), HeldToday: []string{"api"}},
|
||||
map[string]string{"lemurs": fp("e"), "web": fp("1"), "api": fp("2"), "spiegel": fp("3")}, nil, "p-h1", 3)
|
||||
if part.Verdict != delivery.CandidateRefused || len(part.Refusals) != 2 || len(part.Held) != 1 || part.Held[0].Module != "api" {
|
||||
t.Errorf("held as today in part: %+v", part)
|
||||
}
|
||||
|
||||
// The fold: spiegel runs inside the node's own composition at a build of the old form. Its resources are carried as
|
||||
// they run, each after the one it runs after: the node's own is the one it runs, and the node composes itself.
|
||||
|
||||
@@ -124,6 +124,10 @@ type Scope struct {
|
||||
Placements map[string]*delivery.Placement
|
||||
InOwn []string
|
||||
Running []delivery.Composed
|
||||
// HeldToday is, in an answer, the modules the node holds today ("node" for its own composition): their have's phase
|
||||
// is held, or they are held inside its own composition. A candidate that holds one too holds it as today, which is
|
||||
// said under CandidateAnswer.Held and is no refusal (issue 1126).
|
||||
HeldToday []string
|
||||
// SameValue, in an answer, says whether two sealed values open to the same value (the node opens them with its own
|
||||
// key), so a differing resource shows a value sealed again as such (delivery.CompareComposedWith). Nil shows them
|
||||
// as differing.
|
||||
|
||||
+129
-4
@@ -5,9 +5,11 @@
|
||||
package judge
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -121,7 +123,7 @@ func Intended(current, laidInput *compose.Input, currentPlaced map[string]delive
|
||||
// shown in the ask's brief. The seed alone fails on any difference.
|
||||
func JudgeAnswers(p delivery.PlacementProposal, intended map[string]map[string]bool, answers map[string]*delivery.CandidateAnswer,
|
||||
revisions map[string]uint64, online map[string]bool, pulsing map[string]bool, now time.Time) delivery.PlacementVerdict {
|
||||
v := delivery.PlacementVerdict{At: now.UTC(), Intended: map[string][]string{}, Unintended: map[string][]string{},
|
||||
v := delivery.PlacementVerdict{At: now.UTC(), Intended: map[string][]string{}, Unintended: map[string][]string{}, Carried: map[string][]string{},
|
||||
Answers: map[string]delivery.CandidateAnswer{}}
|
||||
for node, set := range intended {
|
||||
for m := range set {
|
||||
@@ -150,7 +152,8 @@ func JudgeAnswers(p delivery.PlacementProposal, intended map[string]map[string]b
|
||||
want := intended[node]
|
||||
differs := func(module, words string) {
|
||||
if p.Seed {
|
||||
v.Fails = append(v.Fails, fmt.Sprintf("%s %s with the seed, which must be the same as what runs", node, words))
|
||||
// The seed carries what its nodes found, each named in the brief for the operator who approves it.
|
||||
v.Carried[node] = append(v.Carried[node], words)
|
||||
return
|
||||
}
|
||||
if !want[module] {
|
||||
@@ -158,7 +161,7 @@ func JudgeAnswers(p delivery.PlacementProposal, intended map[string]map[string]b
|
||||
}
|
||||
}
|
||||
for _, d := range a.Differs {
|
||||
differs(d.Module, fmt.Sprintf("composes %s differently (%s)", d.Module, d.Why))
|
||||
differs(d.Module, differenceWords(d))
|
||||
}
|
||||
for _, d := range a.Added {
|
||||
differs(d.Module, "would run "+d.Module)
|
||||
@@ -174,6 +177,9 @@ func JudgeAnswers(p delivery.PlacementProposal, intended map[string]map[string]b
|
||||
if len(v.Unintended) == 0 {
|
||||
v.Unintended = nil
|
||||
}
|
||||
if len(v.Carried) == 0 {
|
||||
v.Carried = nil
|
||||
}
|
||||
if len(v.Answers) == 0 {
|
||||
v.Answers = nil
|
||||
}
|
||||
@@ -209,7 +215,7 @@ func JudgeAnswers(p delivery.PlacementProposal, intended map[string]map[string]b
|
||||
}
|
||||
|
||||
// Brief is the verdict as the ask's brief shows it, by node (§5): what the proposal changes and what each node answered
|
||||
// beside it, marked, with the proposal's why.
|
||||
// beside it, marked, and what stays held as the node holds it today, with the proposal's why.
|
||||
func Brief(p delivery.PlacementProposal, v delivery.PlacementVerdict) string {
|
||||
var b strings.Builder
|
||||
fmt.Fprintf(&b, "%s (proposed by %s)", p.Why, p.By)
|
||||
@@ -223,6 +229,14 @@ func Brief(p delivery.PlacementProposal, v delivery.PlacementVerdict) string {
|
||||
for _, n := range p.Nodes() {
|
||||
nodes[n] = true
|
||||
}
|
||||
for n, a := range v.Answers {
|
||||
if len(a.Held) > 0 {
|
||||
nodes[n] = true
|
||||
}
|
||||
}
|
||||
for n := range v.Carried {
|
||||
nodes[n] = true
|
||||
}
|
||||
names := make([]string, 0, len(nodes))
|
||||
for n := range nodes {
|
||||
names = append(names, n)
|
||||
@@ -238,6 +252,117 @@ func Brief(p delivery.PlacementProposal, v delivery.PlacementVerdict) string {
|
||||
if m := v.Unintended[n]; len(m) > 0 {
|
||||
fmt.Fprintf(&b, "; also differs, not computed: %s", strings.Join(m, ", "))
|
||||
}
|
||||
if m := v.Carried[n]; len(m) > 0 {
|
||||
fmt.Fprintf(&b, "; the seed carries: %s", strings.Join(m, "; "))
|
||||
}
|
||||
if held := v.Answers[n].Held; len(held) > 0 {
|
||||
var names []string
|
||||
for _, h := range held {
|
||||
names = append(names, h.Module)
|
||||
}
|
||||
fmt.Fprintf(&b, "; stays held as today: %s", strings.Join(names, ", "))
|
||||
}
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// differenceWords names one module's difference: the module, the first field that differs in it with its value as it
|
||||
// runs and as the seed composes it, each cut to about 60 characters ("(value cut)" where a body shown is cut before the
|
||||
// field), and how many fields more; or the resource and why when no body is shown.
|
||||
func differenceWords(d delivery.ModuleDiffers) string {
|
||||
for _, b := range d.Bodies {
|
||||
if len(b.Paths) == 0 {
|
||||
continue
|
||||
}
|
||||
path := b.Paths[0]
|
||||
words := fmt.Sprintf("%s: %s %s → %s", d.Module, fieldName(b.Running, b.Composed, path),
|
||||
valueAt(b.Running, path), valueAt(b.Composed, path))
|
||||
if n := len(b.Paths) - 1; n > 0 {
|
||||
words += fmt.Sprintf(" (and %d more fields)", n)
|
||||
}
|
||||
return words
|
||||
}
|
||||
if d.Resource != "" {
|
||||
return fmt.Sprintf("%s: %s (%s)", d.Module, d.Resource, d.Why)
|
||||
}
|
||||
return fmt.Sprintf("%s: %s", d.Module, d.Why)
|
||||
}
|
||||
|
||||
// shownValue bounds a value named in a brief.
|
||||
const shownValue = 60
|
||||
|
||||
// walk follows a differing path (delivery.DifferingPaths: keys joined by dots, a key may hold dots, and a JSON value
|
||||
// written in a string is read as JSON) through a body, answering the keys it passed and the value at its end.
|
||||
func walk(body, path string) (keys []string, value any, ok bool) {
|
||||
var v any
|
||||
if json.Unmarshal([]byte(body), &v) != nil {
|
||||
return nil, nil, false
|
||||
}
|
||||
rest := path
|
||||
for rest != "" {
|
||||
if s, isString := v.(string); isString {
|
||||
var inner any
|
||||
if json.Unmarshal([]byte(s), &inner) != nil {
|
||||
return nil, nil, false
|
||||
}
|
||||
v = inner
|
||||
}
|
||||
switch t := v.(type) {
|
||||
case map[string]any:
|
||||
best := ""
|
||||
for k := range t {
|
||||
if (rest == k || strings.HasPrefix(rest, k+".")) && len(k) > len(best) {
|
||||
best = k
|
||||
}
|
||||
}
|
||||
if best == "" {
|
||||
return nil, nil, false
|
||||
}
|
||||
keys, v, rest = append(keys, best), t[best], strings.TrimPrefix(strings.TrimPrefix(rest, best), ".")
|
||||
case []any:
|
||||
seg, after, _ := strings.Cut(rest, ".")
|
||||
n, err := strconv.Atoi(seg)
|
||||
if err != nil || n < 0 || n >= len(t) {
|
||||
return nil, nil, false
|
||||
}
|
||||
keys, v, rest = append(keys, seg), t[n], after
|
||||
default:
|
||||
return nil, nil, false
|
||||
}
|
||||
}
|
||||
return keys, v, true
|
||||
}
|
||||
|
||||
// valueAt is the value at path in a shown body, cut to shownValue, or why it is not shown.
|
||||
func valueAt(body, path string) string {
|
||||
if body == "" {
|
||||
return "(none)"
|
||||
}
|
||||
_, v, ok := walk(body, path)
|
||||
if !ok {
|
||||
return "(value cut)"
|
||||
}
|
||||
s, isString := v.(string)
|
||||
if !isString {
|
||||
raw, _ := json.Marshal(v)
|
||||
s = string(raw)
|
||||
}
|
||||
if r := []rune(s); len(r) > shownValue {
|
||||
s = string(r[:shownValue-1]) + "…"
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// fieldName is the differing field as a person reads it: its parent's name and its own, `PARENT[key]`, from whichever
|
||||
// body reaches it; the path's end when neither does.
|
||||
func fieldName(running, composed, path string) string {
|
||||
for _, body := range []string{running, composed} {
|
||||
if keys, _, ok := walk(body, path); ok && len(keys) > 0 {
|
||||
if len(keys) == 1 {
|
||||
return keys[0]
|
||||
}
|
||||
return keys[len(keys)-2] + "[" + keys[len(keys)-1] + "]"
|
||||
}
|
||||
}
|
||||
return path
|
||||
}
|
||||
|
||||
@@ -47,11 +47,17 @@ func TestJudge(t *testing.T) {
|
||||
{"a module added where intended", false, map[string]*delivery.CandidateAnswer{"anchor": {Proposal: "p-one", Revision: 7, Verdict: delivery.CandidateDiffers,
|
||||
Added: []delivery.ModuleAdded{{Module: "web", Unissued: []string{"its own secret"}}}}, "laptop": same(8)}, nil, delivery.VerdictPass, "", ""},
|
||||
{"a difference not computed passes, marked", false, map[string]*delivery.CandidateAnswer{"anchor": differs(7, "mail"), "laptop": same(8)}, nil, delivery.VerdictPass, "", "mail"},
|
||||
{"the seed with a difference", true, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": differs(8, "web")}, nil, delivery.VerdictFail, "laptop composes web differently", ""},
|
||||
{"the seed carries a difference, named", true, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": {Proposal: "p-one", Revision: 8,
|
||||
Verdict: delivery.CandidateDiffers, Differs: []delivery.ModuleDiffers{{Module: "node-tools", Resource: "node-tools.runtime", Why: "the resource differs",
|
||||
Bodies: []delivery.ResourceBodies{{ID: "node-tools.runtime", Paths: []string{"env.MESH_OWNED_PATHS./etc/resolv.conf", "env.MESH_TOOL_MODULES"},
|
||||
Running: `{"env":{"MESH_OWNED_PATHS":"{\"/etc/hosts\":\"hostname\",\"/etc/resolv.conf\":\"networkmanager (node-uplink)\"}"}}`,
|
||||
Composed: `{"env":{"MESH_OWNED_PATHS":"{\"/etc/hosts\":\"hostname\",\"/etc/resolv.conf\":\"systemd-resolved (node-resolver)\"}","MESH_TOOL_MODULES":"… (cut at 8192 bytes)`}}}}}}, nil, delivery.VerdictPass, "", ""},
|
||||
{"a refusal", false, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": {Proposal: "p-one", Revision: 8, Verdict: delivery.CandidateRefused,
|
||||
Refusals: []delivery.ModuleRefused{{Module: "web", Why: "no own secret"}}}}, nil, delivery.VerdictFail, "laptop refuses", ""},
|
||||
{"a node not answering, within its bound", false, map[string]*delivery.CandidateAnswer{"anchor": same(7)}, map[string]bool{"laptop": true}, delivery.VerdictWaits, "laptop (offline)", ""},
|
||||
{"an answer of an earlier revision, past its bound", false, map[string]*delivery.CandidateAnswer{"anchor": same(6), "laptop": same(8)}, nil, delivery.VerdictFail, "anchor did not answer", ""},
|
||||
{"a module held as today passes, shown in the brief", false, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": {Proposal: "p-one", Revision: 8,
|
||||
Verdict: delivery.CandidateSame, Held: []delivery.ModuleRefused{{Module: "spiegel", Why: "a required value is unset"}}}}, nil, delivery.VerdictPass, "", ""},
|
||||
{"its later answer", false, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": same(8)}, nil, delivery.VerdictPass, "", ""},
|
||||
} {
|
||||
q := p
|
||||
@@ -61,6 +67,21 @@ func TestJudge(t *testing.T) {
|
||||
if v.State != row.state || !strings.Contains(all, row.words) {
|
||||
t.Errorf("%s: %s, %q", row.name, v.State, all)
|
||||
}
|
||||
if row.seed && row.state == delivery.VerdictPass && !strings.Contains(Brief(q, v), "the seed carries: node-tools: MESH_OWNED_PATHS[/etc/resolv.conf] networkmanager (node-uplink) → (value cut) (and 1 more fields)") {
|
||||
t.Errorf("%s: the brief does not name it: %s", row.name, Brief(q, v))
|
||||
}
|
||||
if row.seed && row.state == delivery.VerdictPass {
|
||||
whole := row.answers["laptop"].Differs[0]
|
||||
whole.Bodies = []delivery.ResourceBodies{{ID: "r", Paths: []string{"env.MESH_OWNED_PATHS./etc/resolv.conf"},
|
||||
Running: `{"env":{"MESH_OWNED_PATHS":"{\"/etc/resolv.conf\":\"networkmanager (node-uplink)\"}"}}`,
|
||||
Composed: `{"env":{"MESH_OWNED_PATHS":"{\"/etc/resolv.conf\":\"systemd-resolved (node-resolver)\"}"}}`}}
|
||||
if got := differenceWords(whole); got != "node-tools: MESH_OWNED_PATHS[/etc/resolv.conf] networkmanager (node-uplink) → systemd-resolved (node-resolver)" {
|
||||
t.Errorf("the difference is named %q", got)
|
||||
}
|
||||
}
|
||||
if strings.HasPrefix(row.name, "a module held as today") && !strings.Contains(Brief(q, v), "stays held as today: spiegel") {
|
||||
t.Errorf("%s: the brief does not show it: %s", row.name, Brief(q, v))
|
||||
}
|
||||
if row.marked != "" && (!slices.Contains(v.Unintended["anchor"], row.marked) || !strings.Contains(Brief(q, v), "not computed: "+row.marked)) {
|
||||
t.Errorf("%s: the difference is not marked: %+v, %s", row.name, v.Unintended, Brief(q, v))
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ package delivery
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
)
|
||||
|
||||
@@ -121,6 +122,9 @@ type CandidateAnswer struct {
|
||||
Added []ModuleAdded `json:"added,omitempty"`
|
||||
Removed []string `json:"removed,omitempty"`
|
||||
Refusals []ModuleRefused `json:"refusals,omitempty"`
|
||||
// Held is each module the candidate cannot compose that the node holds today too, with why: held as today, which
|
||||
// changes nothing and counts for no verdict (issue 1126: a seed reproduces today).
|
||||
Held []ModuleRefused `json:"held,omitempty"`
|
||||
// Refused is why the node could not compose with the candidate at all; Refusals is then empty.
|
||||
Refused string `json:"refused,omitempty"`
|
||||
// Returns is, for a proposal that takes the node's placement key away, what the node does then: it returns to the
|
||||
@@ -233,7 +237,7 @@ func (a CandidateAnswer) check() error {
|
||||
return fmt.Errorf("delivery: a module the candidate adds is not named")
|
||||
}
|
||||
}
|
||||
for _, d := range a.Refusals {
|
||||
for _, d := range append(slices.Clone(a.Refusals), a.Held...) {
|
||||
if d.Module == "" || d.Why == "" {
|
||||
return fmt.Errorf("delivery: a candidate refusal names no module or no why")
|
||||
}
|
||||
|
||||
@@ -356,15 +356,19 @@ func NewProposalID(at time.Time) string {
|
||||
// is shown by node: Intended is what the holder computed the proposal changes, Unintended what a node answered beside
|
||||
// it, marked as such in the brief.
|
||||
type PlacementVerdict struct {
|
||||
State string `json:"state"`
|
||||
At time.Time `json:"at"`
|
||||
Fails []string `json:"fails,omitempty"`
|
||||
Silent []string `json:"silent,omitempty"`
|
||||
Until time.Time `json:"until,omitempty"`
|
||||
Intended map[string][]string `json:"intended,omitempty"`
|
||||
Unintended map[string][]string `json:"unintended,omitempty"`
|
||||
Answers map[string]CandidateAnswer `json:"answers,omitempty"`
|
||||
Notes []string `json:"notes,omitempty"`
|
||||
State string `json:"state"`
|
||||
At time.Time `json:"at"`
|
||||
Fails []string `json:"fails,omitempty"`
|
||||
Silent []string `json:"silent,omitempty"`
|
||||
Until time.Time `json:"until,omitempty"`
|
||||
Intended map[string][]string `json:"intended,omitempty"`
|
||||
Unintended map[string][]string `json:"unintended,omitempty"`
|
||||
// Carried is, for the seed, by node each difference a node found that the seed carries, named for the operator who
|
||||
// approves it knowingly ("<module>: <resource> at <path>"): the seed fails on no difference, only on a refusal or
|
||||
// silence (issue 1126).
|
||||
Carried map[string][]string `json:"carried,omitempty"`
|
||||
Answers map[string]CandidateAnswer `json:"answers,omitempty"`
|
||||
Notes []string `json:"notes,omitempty"`
|
||||
}
|
||||
|
||||
// PlacementProposal is an open proposal, the value of its key in PlacementProposalsBucket.
|
||||
|
||||
Reference in New Issue
Block a user