Merge pull request 'disk-load watches the node-engine's data items instead of composing them, and each round says how far it got (ADR 0577)' (#584) from feat/data-items into main

modules/disk-load/.gitignore

@@ -1 +1,2 @@
-disk-load
+/disk-load
+/cmd/disk-load/disk-load

modules/disk-load/cmd/disk-load/data.go

@@ -0,0 +1,123 @@
+package main
+
+// Where disk-load learns the data items it measures (novox/hq ADR 0577 clause 2): the node-engine's state data
+// (delivery.EngineDataBucket), one key `<node>.<module>` per module placed on this node that declares data items, as the
+// engine composed them. disk-load watches its own node's keys and composes nothing: it reads no resolved values, no
+// manifests and no placement, so a manifest its mesh-sdk cannot judge never silences it, and a round reads one watched
+// map instead of several hundred keys over the tunnel (issue 1955).
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// Watched is this node's data items as the engine's state data holds them, by module, kept from a watch of its keys.
+type Watched struct {
+	Node string
+
+	mu       sync.Mutex
+	byKey    map[string]string // key to its module, for a delete that carries no value
+	byModule map[string][]catalogue.PlacedDataItem
+	ready    bool
+}
+
+// ErrNotWatchedYet is a round asked before the watch handed over the engine's current keys: no item is known yet, and
+// none is taken for gone.
+var ErrNotWatchedYet = errors.New("the node-engine's state " + delivery.EngineDataBucket + " is not watched yet: " +
+	"no data item of this machine is known")
+
+// Take keeps one change of a key and answers the module it is of and whether its items changed.
+func (w *Watched) Take(ch stdio.StateChange) (string, bool, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if w.byKey == nil {
+		w.byKey, w.byModule = map[string]string{}, map[string][]catalogue.PlacedDataItem{}
+	}
+	if ch.Op != stdio.OpPut {
+		module, known := w.byKey[ch.Key]
+		if !known {
+			return "", false, nil
+		}
+		delete(w.byKey, ch.Key)
+		delete(w.byModule, module)
+		return module, true, nil
+	}
+	d, err := delivery.DecodeEngineData(ch.Value)
+	if err != nil {
+		return "", false, fmt.Errorf("%s: %w", ch.Key, err)
+	}
+	if d.Node != w.Node {
+		return "", false, fmt.Errorf("%s names the machine %q, and this is %s", ch.Key, d.Node, w.Node)
+	}
+	for _, it := range d.Items {
+		if it.Module != d.Module {
+			return "", false, fmt.Errorf("%s holds an item of %q, and is %s's", ch.Key, it.Module, d.Module)
+		}
+	}
+	changed := !sameItems(w.byModule[d.Module], d.Items)
+	w.byKey[ch.Key] = d.Module
+	w.byModule[d.Module] = d.Items
+	return d.Module, changed, nil
+}
+
+// Ready says the watch handed over every current key: from now on an item missing is an item gone.
+func (w *Watched) Ready() {
+	w.mu.Lock()
+	w.ready = true
+	w.mu.Unlock()
+}
+
+// All is every module's items, as the Keeper reads them; refused until the watch is ready.
+func (w *Watched) All(context.Context) (map[string][]catalogue.PlacedDataItem, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if !w.ready {
+		return nil, ErrNotWatchedYet
+	}
+	out := make(map[string][]catalogue.PlacedDataItem, len(w.byModule))
+	for m, its := range w.byModule {
+		out[m] = append([]catalogue.PlacedDataItem(nil), its...)
+	}
+	return out, nil
+}
+
+func sameItems(a, b []catalogue.PlacedDataItem) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+// Follow watches this node's keys of the engine's state data, and asks the keeper to measure a module whose items changed
+// after the watch began; it returns once the current keys were handed over (stdio.KeptState.Follow).
+func (w *Watched) Follow(k *Keeper) error {
+	pattern, err := delivery.EngineDataNodeKeys(w.Node)
+	if err != nil {
+		return err
+	}
+	stdio.State(delivery.EngineDataRead).Follow(pattern, func(ch stdio.StateChange) error {
+		module, changed, err := w.Take(ch)
+		if err != nil || !changed || ch.Current {
+			return err
+		}
+		k.Due(module)
+		return nil
+	})
+	w.Ready()
+	// The handed-over set is complete now: a module whose key went while the watch was down is retired once.
+	k.RetireUnkeyed(context.Background())
+	// A round refused while the watch caught up runs now, not at the next hour (the delivery's readings within minutes).
+	go k.RoundIfRefused(context.Background())
+	return nil
+}

modules/disk-load/cmd/disk-load/data_test.go

@@ -0,0 +1,190 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/health"
+)
+
+// ADR 0577 clauses 2, 3 (and the disk-load check of clause 5): the items measured come from the node-engine's keys of
+// this machine as the watch hands them over, and none is known (none taken for gone) before the watch caught up; a key
+// that changes after the watch began has its module measured again, and one deleted is measured no more; an item whose
+// measurement fails is a failed item of the round, which measures the rest, and the round's key says so.
+func TestTheItemsComeFromTheEnginesKeysAndARoundSaysHowFarItGot(t *testing.T) {
+	key := func(module string) string { k, _ := delivery.EngineDataKey("ace", module); return k }
+	value := func(module string, items ...catalogue.PlacedDataItem) json.RawMessage {
+		raw, err := delivery.EncodeEngineData(delivery.EngineData{Node: "ace", Module: module, Digest: "sha256:" + module, Items: items})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return raw
+	}
+	pg := catalogue.PlacedDataItem{Module: "pg", ID: "data", Class: "valuable", Path: "/srv/pg", Measure: catalogue.MeasureShallow}
+	gone := catalogue.PlacedDataItem{Module: "pg", ID: "wal", Class: "valuable", Path: "/srv/missing", Measure: catalogue.MeasureShallow}
+	web := catalogue.PlacedDataItem{Module: "web", ID: "uploads", Class: "valuable", Path: "/srv/web", Measure: catalogue.MeasureShallow}
+	measured := map[string]int{}
+	run := func(_ context.Context, name string, args ...string) (string, error) {
+		switch {
+		case name == "test" && args[1] == "/srv/missing":
+			return "", errors.New("no such file")
+		case name == "test":
+			return "", nil
+		case name == "bash" && strings.Contains(args[1], "-maxdepth 1"):
+			measured[strings.Split(args[1], "'")[1]]++
+			return "1760000000 2\n", nil
+		}
+		return "", os.ErrNotExist
+	}
+	now := time.Date(2026, 11, 1, 0, 30, 0, 0, time.UTC)
+	readings, rounds := &memStore{m: map[string]json.RawMessage{}}, &memStore{m: map[string]json.RawMessage{}}
+	w := &Watched{Node: "ace"}
+	var said []string
+	k := &Keeper{Node: "ace", Items: w.All, Store: readings, Rounds: rounds,
+		Say:      func(f string, a ...any) { said = append(said, f) },
+		Measurer: Measurer{Run: run, Now: func() time.Time { return now }}}
+	round := func() Round {
+		var r Round
+		raw, _ := rounds.Get("ace")
+		if err := json.Unmarshal(raw, &r); err != nil {
+			t.Fatalf("the round's key: %v", err)
+		}
+		return r
+	}
+
+	// Before the watch caught up nothing is known, and a round says so in its key rather than retiring every reading.
+	if _, _, err := k.Round(context.Background(), false); !errors.Is(err, ErrNotWatchedYet) {
+		t.Fatalf("a round before the watch: %v", err)
+	}
+	if r := round(); r.Ended == nil || !strings.Contains(r.Error, "not watched yet") {
+		t.Fatalf("the round before the watch is kept as %+v", r)
+	}
+	refusedAt := round().Started
+	for _, step := range []struct {
+		name    string
+		change  stdio.StateChange
+		module  string
+		changed bool
+		err     bool
+	}{
+		{"a current key", stdio.StateChange{Key: key("pg"), Op: stdio.OpPut, Value: value("pg", pg, gone), Current: true}, "pg", true, false},
+		{"another current key", stdio.StateChange{Key: key("web"), Op: stdio.OpPut, Value: value("web", web), Current: true}, "web", true, false},
+		{"the same items again", stdio.StateChange{Key: key("web"), Op: stdio.OpPut, Value: value("web", web)}, "web", false, false},
+		{"another machine's items", stdio.StateChange{Key: key("x"), Op: stdio.OpPut, Value: json.RawMessage(
+			`{"v":1,"node":"g14","module":"x","digest":"d","items":[]}`)}, "", false, true},
+		{"a delete of a key never seen", stdio.StateChange{Key: key("y"), Op: stdio.OpDelete}, "", false, false},
+	} {
+		module, changed, err := w.Take(step.change)
+		if module != step.module || changed != step.changed || (err != nil) != step.err {
+			t.Errorf("%s: taken as %q, changed %v, %v", step.name, module, changed, err)
+		}
+	}
+	w.Ready()
+
+	// The round refused before the watch caught up runs once it has (Watched.Follow), not at the next hour: it measures
+	// every item it was handed; the missing one fails, and the round goes on and says it.
+	now = now.Add(time.Minute)
+	if !k.RoundIfRefused(context.Background()) || k.RoundIfRefused(context.Background()) {
+		t.Fatal("the round refused before the watch was not run once when the watch caught up")
+	}
+	r := round()
+	if !r.Started.After(refusedAt) {
+		t.Fatalf("the round's key still says the refused round: %+v", r)
+	}
+	if r.Ended == nil || r.Items != 3 || r.Measured != 3 || len(r.Failed) != 1 || r.Failed[0].Item != "pg.wal" ||
+		!strings.Contains(r.Failed[0].Why, "does not exist") || measured["/srv/pg"] != 1 || measured["/srv/web"] != 1 {
+		t.Fatalf("the round's key says %+v; measured %v", r, measured)
+	}
+	if !strings.Contains(strings.Join(said, "\n"), "the round ended") || !strings.Contains(strings.Join(said, "\n"), "starts") {
+		t.Errorf("the journal said %v", said)
+	}
+
+	// A key that changes after the watch began has its module measured again, at once, by Work.
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go k.Work(ctx)
+	moved := web
+	moved.Path = "/srv/web2"
+	if module, changed, err := w.Take(stdio.StateChange{Key: key("web"), Op: stdio.OpPut, Value: value("web", moved)}); err != nil || !changed {
+		t.Fatalf("a changed key: %s %v %v", module, changed, err)
+	}
+	k.Due("web")
+	waitFor(t, func() bool {
+		raw := readingOf(readings, "ace.web.uploads")
+		return raw != nil && raw.Path == "/srv/web2"
+	})
+	// A key deleted: its module's items are measured no more, and its reading is retired.
+	if module, changed, _ := w.Take(stdio.StateChange{Key: key("web"), Op: stdio.OpDelete}); module != "web" || !changed {
+		t.Fatalf("a deleted key was taken as %q, %v", module, changed)
+	}
+	k.Due("web")
+	waitFor(t, func() bool { r := readingOf(readings, "ace.web.uploads"); return r != nil && !r.RetiredAt.IsZero() })
+	before := measured["/srv/web2"]
+	if _, _, err := k.Round(context.Background(), false); err != nil {
+		t.Fatal(err)
+	}
+	if measured["/srv/web2"] != before || round().Items != 2 {
+		t.Errorf("an item whose key was deleted was measured again: %d times, round %+v", measured["/srv/web2"]-before, round())
+	}
+
+	// A restart: oldie's key was deleted while the watch was down, so no delete arrives. Once the watch handed over its
+	// keys, oldie's reading is retired and pg's kept; an empty set retires nothing (the engine has not written yet).
+	_ = readings.Put("ace.oldie.x", health.Reading{V: health.ReadingVersion, Node: "ace",
+		PlacedDataItem: catalogue.PlacedDataItem{Module: "oldie", ID: "x", Path: "/srv/oldie"}})
+	for _, restart := range []struct {
+		name    string
+		keys    []stdio.StateChange
+		retired bool
+	}{
+		{"no key yet", nil, false},
+		{"pg's key alone", []stdio.StateChange{{Key: key("pg"), Op: stdio.OpPut, Value: value("pg", pg, gone), Current: true}}, true},
+	} {
+		w2 := &Watched{Node: "ace"}
+		said = nil
+		k2 := &Keeper{Node: "ace", Items: w2.All, Store: readings, Say: k.Say, Measurer: k.Measurer}
+		for _, ch := range restart.keys {
+			if _, _, err := w2.Take(ch); err != nil {
+				t.Fatal(err)
+			}
+		}
+		w2.Ready()
+		k2.RetireUnkeyed(context.Background())
+		oldie, pgData := readingOf(readings, "ace.oldie.x"), readingOf(readings, "ace.pg.data")
+		if oldie == nil || pgData == nil || oldie.RetiredAt.IsZero() == restart.retired || !pgData.RetiredAt.IsZero() {
+			t.Errorf("%s: oldie %+v, pg %+v; oldie retired: want %v, pg never", restart.name, oldie, pgData, restart.retired)
+		}
+		if !restart.retired && !strings.Contains(strings.Join(said, "\n"), "holds no key") {
+			t.Errorf("%s: the skip was not said: %v", restart.name, said)
+		}
+	}
+}
+
+func readingOf(s *memStore, key string) *health.Reading {
+	raw, _ := s.Get(key)
+	if raw == nil {
+		return nil
+	}
+	r, err := health.DecodeReading(raw)
+	if err != nil {
+		return nil
+	}
+	return &r
+}
+
+func waitFor(t *testing.T, done func() bool) {
+	t.Helper()
+	for end := time.Now().Add(5 * time.Second); time.Now().Before(end); time.Sleep(5 * time.Millisecond) {
+		if done() {
+			return
+		}
+	}
+	t.Fatal("not done within 5 s")
+}

modules/disk-load/cmd/disk-load/keeper.go

@@ -7,12 +7,14 @@ import (
 	"errors"
 	"fmt"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/health"
 	"os"
 	"os/exec"
 	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
@@ -25,47 +27,130 @@ type Store interface {
 	Keys() ([]string, error)
 }
 
-// busStore is the state through the runtime.
+// busStore is one of disk-load's states through the runtime: readings, or rounds.
-type busStore struct{}
+type busStore struct{ state string }
 
-func (busStore) Get(key string) (json.RawMessage, error) {
+func (b busStore) Get(key string) (json.RawMessage, error) {
-	e, err := stdio.State(StateReadings).Get(key)
+	e, err := stdio.State(b.state).Get(key)
 	if err != nil || e == nil {
 		return nil, err
 	}
 	return e.Value, nil
 }
-func (busStore) Put(key string, v any) error {
+func (b busStore) Put(key string, v any) error {
-	_, err := stdio.State(StateReadings).Put(key, v)
+	_, err := stdio.State(b.state).Put(key, v)
 	return err
 }
-func (busStore) Keys() ([]string, error) { return stdio.State(StateReadings).Keys() }
+func (b busStore) Keys() ([]string, error) { return stdio.State(b.state).Keys() }
 
 // Keeper measures the items of its node and keeps the readings.
 type Keeper struct {
 	Node string
-	// Items is every data item the modules placed on this node declare, composed from the bus (frombus.DataItems).
+	// Items is every data item the modules placed on this node declare, as the node-engine composed them and its state
+	// data holds them (Watched.All, novox/hq ADR 0577 clause 2).
 	Items    func(ctx context.Context) (map[string][]catalogue.PlacedDataItem, error)
 	Measurer Measurer
 	Store    Store
-	Say      func(string, ...any)
+	// Rounds is the state rounds, where each round says how far it got (ADR 0577 clause 3); nil keeps none.
+	Rounds Store
+	Say    func(string, ...any)
 
 	mu sync.Mutex // one round at a time
+
+	// pending are the modules whose items changed since they were last measured (Due), measured by Work.
+	pmu     sync.Mutex
+	pending map[string]bool
+	wake    chan struct{}
+
+	// unwatched says a round was refused because the watch had not caught up (ErrNotWatchedYet): one runs as soon as
+	// it has (Watched.Follow), not an hour later.
+	unwatched atomic.Bool
+}
+
+// RoundIfRefused runs one round now when the last was refused for a watch not caught up, and says whether it did.
+func (k *Keeper) RoundIfRefused(ctx context.Context) bool {
+	if !k.unwatched.CompareAndSwap(true, false) {
+		return false
+	}
+	_, _, _ = k.Round(ctx, false)
+	return true
+}
+
+// SayEvery is how many items a round measures between two lines saying how far it got.
+const SayEvery = 50
+
+// RoundVersion is the version of a round's layout.
+const RoundVersion = 1
+
+// Round is the value of a node's key of the state rounds: when its last round started and ended, how many items it
+// had to measure, how many it measured, wrote and kept, and each item that failed and why. Ended is nil while it runs;
+// Error says why a round measured nothing at all.
+type Round struct {
+	V        int          `json:"v"`
+	Node     string       `json:"node"`
+	Walks    bool         `json:"walks,omitempty"`
+	Started  time.Time    `json:"started"`
+	Ended    *time.Time   `json:"ended,omitempty"`
+	Items    int          `json:"items"`
+	Measured int          `json:"measured"`
+	Written  int          `json:"written"`
+	Kept     int          `json:"kept"`
+	Failed   []FailedItem `json:"failed,omitempty"`
+	Error    string       `json:"error,omitempty"`
+}
+
+// FailedItem is one item a round could not measure or keep, `<module>.<item>`, and why.
+type FailedItem struct {
+	Item string `json:"item"`
+	Why  string `json:"why"`
+}
+
+// putRound writes the round to this node's key; a round that cannot be written is said, and measures on.
+func (k *Keeper) putRound(r Round) {
+	if k.Rounds == nil {
+		return
+	}
+	if err := k.Rounds.Put(k.Node, r); err != nil {
+		k.Say("the round's outcome cannot be kept in the state %s: %v", StateRounds, err)
+	}
 }
 
 // Round measures every item that is not a cache, one at a time, and writes the readings that changed. An item
 // measured by a walk is walked only when walks is true or its last walk is older than WalkEvery: the hourly round
 // reads counters, and a walk runs at most once a day (ADR 0233 §22).
+//
+// A round says how far it got (novox/hq ADR 0577 clause 3): it writes its outcome to this node's key of the state rounds
+// when it starts, every SayEvery items and when it ends, and says the same in the journal. An item that fails is a
+// failed item of the round, never its end.
 func (k *Keeper) Round(ctx context.Context, walks bool) (written, kept int, err error) {
 	k.mu.Lock()
 	defer k.mu.Unlock()
+	r := Round{V: RoundVersion, Node: k.Node, Walks: walks, Started: k.Measurer.Now().UTC()}
+	end := func() {
+		at := k.Measurer.Now().UTC()
+		r.Ended = &at
+		k.putRound(r)
+	}
+	k.putRound(r)
+	k.Say("a round of measuring the data declared here starts (walks: %v)", walks)
 	items, err := k.Items(ctx)
 	if err != nil {
+		if errors.Is(err, ErrNotWatchedYet) {
+			k.unwatched.Store(true)
+		}
+		r.Error = err.Error()
+		end()
+		k.Say("the round measured nothing: %v", err)
 		return 0, 0, err
 	}
 	modules := make([]string, 0, len(items))
 	for m := range items {
 		modules = append(modules, m)
+		for _, it := range items[m] {
+			if it.Class != ClassCache {
+				r.Items++
+			}
+		}
 	}
 	sort.Strings(modules)
 	var failed []string
@@ -74,42 +159,175 @@ func (k *Keeper) Round(ctx context.Context, walks bool) (written, kept int, err
 			if it.Class == ClassCache {
 				continue
 			}
-			w, err := k.measureOne(ctx, it, walks)
+			w, said, err := k.measureOne(ctx, it, walks)
-			if err != nil {
+			r.Measured++
+			switch {
+			case err != nil:
 				failed = append(failed, fmt.Sprintf("%s.%s: %v", module, it.ID, err))
-				continue
+				r.Failed = append(r.Failed, FailedItem{Item: module + "." + it.ID, Why: err.Error()})
+			case said != "":
+				// Measured, and the measurement failed: its reading says so, and the round counts it failed.
+				r.Failed = append(r.Failed, FailedItem{Item: module + "." + it.ID, Why: said})
 			}
-			if w {
+			if err == nil && w {
 				written++
-			} else {
+			} else if err == nil {
 				kept++
 			}
+			r.Written, r.Kept = written, kept
+			if r.Measured%SayEvery == 0 {
+				k.Say("the round has measured %d of %d items: %d written, %d unchanged, %d failed", r.Measured, r.Items,
+					written, kept, len(r.Failed))
+				k.putRound(r)
+			}
 		}
 	}
-	retired, err := k.retireGone(items)
+	// A round retires nothing: a reading is retired only when its module's key says so (MeasureModule), never because
-	written += retired
+	// an item is missing from what the watch holds now, which may be empty or partial.
-	if err != nil {
+	end()
-		failed = append(failed, err.Error())
+	k.Say("the round ended: %d of %d items measured, %d readings written, %d unchanged, %d failed", r.Measured, r.Items,
-	}
+		written, kept, len(r.Failed))
 	if len(failed) > 0 {
 		return written, kept, errors.New("readings not kept: " + strings.Join(failed, "; "))
 	}
 	return written, kept, nil
 }
 
-// retireGone marks retired every reading of this node whose item the node no longer declares — its module was unassigned
+// Due asks Work to measure a module's items: its key of the engine's state data changed (novox/hq ADR 0577 clause 2).
-// there, or is no longer pulled in — with when and why, and keeps it: the record of what the module left behind, until a
+func (k *Keeper) Due(module string) {
-// person deletes it (ADR 0233). It is cleared when the item is declared again (measureOne).
+	k.pmu.Lock()
-func (k *Keeper) retireGone(declared map[string][]catalogue.PlacedDataItem) (int, error) {
+	if k.pending == nil {
+		k.pending, k.wake = map[string]bool{}, make(chan struct{}, 1)
+	}
+	k.pending[module] = true
+	wake := k.wake
+	k.pmu.Unlock()
+	select {
+	case wake <- struct{}{}:
+	default:
+	}
+}
+
+// Work measures each module Due names, one at a time, until ctx ends.
+func (k *Keeper) Work(ctx context.Context) {
+	k.pmu.Lock()
+	if k.pending == nil {
+		k.pending, k.wake = map[string]bool{}, make(chan struct{}, 1)
+	}
+	wake := k.wake
+	k.pmu.Unlock()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-wake:
+		}
+		for {
+			k.pmu.Lock()
+			var module string
+			for m := range k.pending {
+				module = m
+				break
+			}
+			delete(k.pending, module)
+			k.pmu.Unlock()
+			if module == "" {
+				break
+			}
+			k.MeasureModule(ctx, module)
+		}
+	}
+}
+
+// MeasureModule measures one module's items now, as its key of the engine's state data changed, and retires the
+// readings of a module that no longer declares any here. A walk is not repeated within a day for an item that did not
+// change (measureOne).
+func (k *Keeper) MeasureModule(ctx context.Context, module string) {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	items, err := k.Items(ctx)
+	if err != nil {
+		k.Say("%s's data items changed and are not measured: %v", module, err)
+		return
+	}
+	n, err := k.retireGone(module, items[module])
+	if err != nil {
+		k.Say("%s's readings of items it no longer declares here are not retired: %v", module, err)
+	} else if n > 0 {
+		k.Say("%s no longer declares %d item(s) here: their readings are retired, measured no more", module, n)
+	}
+	if len(items[module]) == 0 {
+		return
+	}
+	written, kept := 0, 0
+	var failed []string
+	for _, it := range items[module] {
+		if it.Class == ClassCache {
+			continue
+		}
+		w, said, err := k.measureOne(ctx, it, false)
+		switch {
+		case err != nil:
+			failed = append(failed, it.ID+": "+err.Error())
+		case said != "":
+			failed = append(failed, it.ID+": "+said)
+		case w:
+			written++
+		default:
+			kept++
+		}
+	}
+	k.Say("%s's data items changed and were measured: %d readings written, %d unchanged, %d failed%s", module, written,
+		kept, len(failed), map[bool]string{true: " (" + strings.Join(failed, "; ") + ")", false: ""}[len(failed) > 0])
+}
+
+// retireGone marks retired every reading of this node of module whose item its key of the engine's state data no longer
+// names — the key was deleted (the module was unassigned there, or is no longer pulled in) or names fewer items — with
+// when and why, and keeps it: the record of what the module left behind, until a person deletes it (ADR 0233). It is
+// asked only on a change of that module's key (novox/hq ADR 0577 clause 2), never from a round. It is cleared when the
+// item is declared again (measureOne).
+func (k *Keeper) retireGone(module string, declared []catalogue.PlacedDataItem) (int, error) {
+	return k.retire(func(r health.Reading) bool { return r.Module == module }, declared)
+}
+
+// RetireUnkeyed retires, once the watch handed over every current key, the readings of this node whose module has no
+// key in that set: a key deleted while the watch was not up (disk-load restarted, the machine rebooted or slept) never
+// reaches disk-load as a delete. An empty set is skipped and said: the engine has not written its keys yet, and an
+// empty set would retire every reading.
+func (k *Keeper) RetireUnkeyed(ctx context.Context) {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	items, err := k.Items(ctx)
+	if err != nil {
+		k.Say("the readings of modules with no key are not looked for: %v", err)
+		return
+	}
+	if len(items) == 0 {
+		k.Say("the node-engine's state %s holds no key of this machine yet: no reading is retired for it", delivery.EngineDataBucket)
+		return
+	}
+	var declared []catalogue.PlacedDataItem
+	for _, its := range items {
+		declared = append(declared, its...)
+	}
+	n, err := k.retire(func(r health.Reading) bool { _, keyed := items[r.Module]; return !keyed }, declared)
+	switch {
+	case err != nil:
+		k.Say("the readings of modules with no key are not all retired: %v", err)
+	case n > 0:
+		k.Say("%d reading(s) of modules with no key of this machine retired, measured no more", n)
+	}
+}
+
+// retire marks retired every reading of this node that of picks and declared does not name.
+func (k *Keeper) retire(of func(health.Reading) bool, declared []catalogue.PlacedDataItem) (int, error) {
 	keys, err := k.Store.Keys()
 	if err != nil {
 		return 0, err
 	}
 	declaredNow := map[string]bool{}
-	for module, its := range declared {
+	for _, it := range declared {
-		for _, it := range its {
+		declaredNow[health.ReadingKey(k.Node, it.Module, it.ID)] = true
-			declaredNow[health.ReadingKey(k.Node, module, it.ID)] = true
-		}
 	}
 	now := k.Measurer.Now().UTC()
 	n := 0
@@ -122,7 +340,7 @@ func (k *Keeper) retireGone(declared map[string][]catalogue.PlacedDataItem) (int
 			continue
 		}
 		r, err := health.DecodeReading(raw)
-		if err != nil || r.Node != k.Node || !r.RetiredAt.IsZero() {
+		if err != nil || r.Node != k.Node || !of(r) || !r.RetiredAt.IsZero() {
 			continue
 		}
 		r.RetiredAt = now
@@ -135,12 +353,15 @@ func (k *Keeper) retireGone(declared map[string][]catalogue.PlacedDataItem) (int
 	return n, nil
 }
 
-func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, walks bool) (bool, error) {
+// measureOne measures one item, unless it is a walk done within WalkEvery of an unchanged item, and writes its reading
+// when it changed. It answers whether it wrote the reading, what the measurement said failed ("" when it did not, or
+// when the item was not measured now), and an error when the reading could not be read or kept.
+func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, walks bool) (bool, string, error) {
 	key := health.ReadingKey(k.Node, it.Module, it.ID)
 	var prev *health.Reading
 	raw, err := k.Store.Get(key)
 	if err != nil {
-		return false, err
+		return false, "", err
 	}
 	if len(raw) > 0 {
 		if p, err := health.DecodeReading(raw); err == nil {
@@ -148,8 +369,9 @@ func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, wa
 		}
 	}
 	now := k.Measurer.Now()
-	if prev != nil && prev.RetiredAt.IsZero() && it.Measure == catalogue.MeasureWalk && !walks && prev.MeasuredAt != nil && now.Sub(*prev.MeasuredAt) < WalkEvery {
+	if prev != nil && prev.RetiredAt.IsZero() && prev.PlacedDataItem == it && it.Measure == catalogue.MeasureWalk && !walks &&
-		return false, nil
+		prev.MeasuredAt != nil && now.Sub(*prev.MeasuredAt) < WalkEvery {
+		return false, "", nil
 	}
 	m := k.Measurer.Measure(ctx, it)
 	next, changed := health.NextReading(prev, k.Node, it, m, now)
@@ -158,9 +380,9 @@ func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, wa
 		next.RetiredAt, next.RetiredWhy, changed = time.Time{}, "", true
 	}
 	if !changed {
-		return false, nil
+		return false, m.Error, nil
 	}
-	return true, k.Store.Put(key, next)
+	return true, m.Error, k.Store.Put(key, next)
 }
 
 // MeasureNow measures one item now, walk or not, and writes its reading when it changed.
@@ -178,7 +400,7 @@ func (k *Keeper) MeasureNow(ctx context.Context, module, item string) (health.Re
 		if it.Class == ClassCache {
 			return health.Reading{}, false, fmt.Errorf("%s.%s is a cache, which is listed and never measured", module, item)
 		}
-		w, err := k.measureOne(ctx, it, true)
+		w, _, err := k.measureOne(ctx, it, true)
 		if err != nil {
 			return health.Reading{}, false, err
 		}
@@ -201,12 +423,7 @@ func (k *Keeper) Loop(ctx context.Context, every time.Duration, first time.Durat
 	case <-time.After(first):
 	}
 	for {
-		w, kept, err := k.Round(ctx, false)
+		_, _, _ = k.Round(ctx, false) // it says its outcome itself, in the journal and its key of the state rounds
-		if err != nil {
-			k.Say("measuring the data declared here: %v", err)
-		} else {
-			k.Say("measured the data declared here: %d readings changed, %d unchanged", w, kept)
-		}
 		select {
 		case <-ctx.Done():
 			return

modules/disk-load/cmd/disk-load/main.go

@@ -4,42 +4,52 @@
 // filesystems' space and the processes doing the I/O — and changes nothing of the machine.
 //
 // It also measures the data items the modules of its node declare (ADR 0233) and keeps each reading and the item's
-// seven-day peak as its own state `readings` (ADR 0571 clause 3), composed from the bus (no file).
+// seven-day peak as its own state `readings` (ADR 0571 clause 3). It learns the items from the node-engine's state data,
+// watching its own node's keys, and composes nothing (ADR 0577 clause 2); each round says how far it got in its state
+// `rounds` (clause 3).
 package main
 
 import (
 	"context"
 	"fmt"
 	"os"
-	"strings"
 	"time"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/announce"
-	"git.novox.be/novox/mesh-sdk/go/catalogue"
-	"git.novox.be/novox/mesh-sdk/go/compose/frombus"
-	"git.novox.be/novox/mesh-sdk/go/compose/fromruntime"
 )
 
 func logf(format string, args ...any) { fmt.Fprintf(os.Stderr, "[disk-load] "+format+"\n", args...) }
 
+// Kept is every state disk-load keeps, as it announces them: the readings, readable by restic, nfs-server, mounts and
+// the self-check (their manifests' `reads`; checks/readable holds it), and the rounds, its own tools' alone.
+var Kept = []announce.State{{Name: StateReadings, Readable: true}, {Name: StateRounds}}
+
 func main() {
-	// Readable: restic, nfs-server, mounts and the self-check read it (their manifests' `reads`); checks/readable holds it.
+	if err := stdio.Keeps(Kept...); err != nil {
-	if err := stdio.Keeps(announce.State{Name: StateReadings, Readable: true}); err != nil {
 		logf("%v", err)
 		os.Exit(1)
 	}
 	node := os.Getenv("MESH_NODE")
-	k := &Keeper{Node: node, Store: busStore{}, Say: logf, Measurer: Measurer{Run: RootRunner, Now: time.Now},
+	watched := &Watched{Node: node}
-		Items: func(ctx context.Context) (map[string][]catalogue.PlacedDataItem, error) {
+	k := &Keeper{Node: node, Store: busStore{StateReadings}, Rounds: busStore{StateRounds}, Say: logf,
-			return placedItems(ctx, node)
+		Measurer: Measurer{Run: RootRunner, Now: time.Now}, Items: watched.All}
-		}}
 	if node != "" {
 		every := time.Hour
 		if d, err := time.ParseDuration(os.Getenv("MESH_DISK_MEASURE_EVERY")); err == nil && d >= time.Minute {
 			every = d
 		}
-		go k.Loop(context.Background(), every, 3*time.Minute)
+		ctx := context.Background()
+		go k.Work(ctx)
+		go func() {
+			<-stdio.Served() // the runtime's channel first: a watch is asked through it
+			if err := watched.Follow(k); err != nil {
+				logf("this machine's keys of the node-engine's data items cannot be watched, and nothing is measured: %v", err)
+				return
+			}
+			logf("watching this machine's keys of the node-engine's data items")
+		}()
+		go k.Loop(ctx, every, 3*time.Minute)
 	}
 	tools := append(Tools(Here()), ReadingTools(k)...)
 	// An empty name serves as the module the runtime names (MESH_SERVED_MODULE): disk-load.
@@ -48,15 +58,3 @@ func main() {
 		os.Exit(1)
 	}
 }
-
-// placedItems is every data item the modules placed on node declare, composed from the bus as the node composes itself.
-func placedItems(ctx context.Context, node string) (map[string][]catalogue.PlacedDataItem, error) {
-	items, unplaced, err := frombus.DataItems(ctx, fromruntime.Reader(), frombus.Mesh{}, frombus.Scope{Node: node})
-	if err != nil {
-		return nil, err
-	}
-	if len(unplaced) > 0 {
-		logf("data items left out as not placeable: %s", strings.Join(unplaced, "; "))
-	}
-	return items, nil
-}

modules/disk-load/cmd/disk-load/readings.go

@@ -8,5 +8,9 @@ import "git.novox.be/novox/mesh-sdk/go/health"
 // StateReadings is the state disk-load keeps.
 const StateReadings = health.ReadingsState
 
+// StateRounds is its state of each node's last round, key `<node>` (novox/hq ADR 0577 clause 3): apart from the readings,
+// so a reader of the readings never meets a round.
+const StateRounds = "rounds"
+
 // ClassCache is the class of an item that is listed and never measured.
 const ClassCache = "cache"

modules/disk-load/cmd/disk-load/readings_test.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"os"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -12,15 +13,26 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/health"
 )
 
-type memStore struct{ m map[string]json.RawMessage }
+type memStore struct {
+	mu sync.Mutex
+	m  map[string]json.RawMessage
+}
 
-func (s *memStore) Get(k string) (json.RawMessage, error) { return s.m[k], nil }
+func (s *memStore) Get(k string) (json.RawMessage, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.m[k], nil
+}
 func (s *memStore) Put(k string, v any) error {
 	raw, _ := json.Marshal(v)
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	s.m[k] = raw
 	return nil
 }
 func (s *memStore) Keys() ([]string, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	var out []string
 	for k := range s.m {
 		out = append(out, k)
@@ -66,11 +78,16 @@ func TestARoundMeasuresTheDeclaredItemsAndWritesWhatChanged(t *testing.T) {
 	if _, _, err := k.MeasureNow(context.Background(), "pg", "data"); err != nil || walks != 2 {
 		t.Fatalf("measure now: %v, %d walks", err, walks)
 	}
-	// An item no longer declared is kept, retired with when and why; declared again, it is not retired any more.
+	// A round retires nothing, even when no item is declared (the watched set may be empty or partial); an item its
+	// module's key no longer names is kept, retired with when and why; declared again, it is not retired any more.
 	delete(placed, "pg")
 	if _, _, err := k.Round(context.Background(), false); err != nil {
 		t.Fatal(err)
 	}
+	if kept, _ := health.DecodeReading(store.m["ace.pg.data"]); !kept.RetiredAt.IsZero() {
+		t.Fatalf("a round retired a reading: %+v", kept)
+	}
+	k.MeasureModule(context.Background(), "pg")
 	gone, _ := health.DecodeReading(store.m["ace.pg.data"])
 	if gone.RetiredAt.IsZero() || !strings.Contains(gone.RetiredWhy, "no longer declared on ace") || gone.FirstSeen.IsZero() {
 		t.Fatalf("a reading of an item no longer declared: %+v", gone)

modules/disk-load/cmd/disk-load/readings_tools.go

@@ -2,9 +2,11 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"sort"
 	"strings"
+	"time"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/health"
@@ -68,6 +70,37 @@ func ReadingTools(k *Keeper) []stdio.Tool {
 					"cap": "none declared: the bus's own limit on a state applies"}}, nil
 			},
 		},
+		{
+			Name: "disk_round",
+			Description: "How far this machine's last round of measuring its data items got (novox/hq ADR 0577 clause 3), from " +
+				"disk-load's state `rounds`, this machine's key `<node>`: when it started and ended (no end while it runs), how " +
+				"many items it had, measured, wrote and kept unchanged, each item that failed and why, and how old the round is. " +
+				"A round that never ends, or fails every item, shows here. (r)",
+			Input: map[string]any{},
+			Run: func(args map[string]any) (any, error) {
+				node := k.Node
+				if k.Rounds == nil {
+					return nil, fmt.Errorf("this disk-load keeps no rounds")
+				}
+				raw, err := k.Rounds.Get(node)
+				if err != nil {
+					return nil, err
+				}
+				if len(raw) == 0 {
+					return nil, fmt.Errorf("no round of %s is kept yet: its first round starts a few minutes after disk-load starts", node)
+				}
+				var r Round
+				if err := json.Unmarshal(raw, &r); err != nil {
+					return nil, fmt.Errorf("the round of %s is not readable: %w", node, err)
+				}
+				state := "ended"
+				if r.Ended == nil {
+					state = "running"
+				}
+				return map[string]any{"round": r, "state": state,
+					"age": k.Measurer.Now().UTC().Sub(r.Started).Round(time.Second).String()}, nil
+			},
+		},
 		{
 			Name: "disk_measure",
 			Description: "Measure one data item of this machine now, whatever its method (a walk too), and write its reading when it " +

modules/disk-load/cmd/disk-load/whole_integration_test.go

@@ -0,0 +1,267 @@
+//go:build integration
+
+package main
+
+// disk-load whole against a real bus (novox/hq ADR 0577 clause 5, the group's integration test): the built binary, played
+// over its real stdin and stdout as the node's tool runtime plays it (mesh/state.* answered from a real nats-server's
+// key-value buckets, a watch handed over from the bus's own watch, mesh/publish onto its subjects). A key of the
+// node-engine's state data written after disk-load started has its module's item read within a minute, and disk-load
+// composes nothing to learn it.
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/health"
+	"git.novox.be/novox/mesh-sdk/go/testbus"
+)
+
+type wire struct {
+	ID     json.RawMessage `json:"id,omitempty"`
+	Method string          `json:"method,omitempty"`
+	Params json.RawMessage `json:"params,omitempty"`
+	Result json.RawMessage `json:"result,omitempty"`
+}
+
+// player is the node's tool runtime, as far as disk-load needs it, over a real bus.
+type player struct {
+	ctx   context.Context
+	nc    *nats.Conn
+	js    jetstream.JetStream
+	mu    sync.Mutex
+	in    io.Writer
+	next  int
+	asked []string // every state call, `<method> <state> <key>`
+}
+
+func (p *player) send(m map[string]any) {
+	raw, _ := json.Marshal(m)
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	_, _ = p.in.Write(append(raw, '\n'))
+}
+
+func (p *player) answer(id json.RawMessage, result any, err error) {
+	if err != nil {
+		p.send(map[string]any{"jsonrpc": "2.0", "id": id, "error": map[string]any{"code": -32000, "message": err.Error()}})
+		return
+	}
+	p.send(map[string]any{"jsonrpc": "2.0", "id": id, "result": result})
+}
+
+// bucket is a state's bucket: another module's `<owner>.<name>`, or disk-load's own by its local name.
+func bucket(state string) string {
+	if owner, name, other := strings.Cut(state, "."); other {
+		return owner + "_" + name
+	}
+	return "disk-load_" + state
+}
+
+func (p *player) serve(out io.Reader) {
+	scanner := bufio.NewScanner(out)
+	scanner.Buffer(make([]byte, 1<<20), 1<<24)
+	for scanner.Scan() {
+		var m wire
+		if json.Unmarshal(scanner.Bytes(), &m) != nil || m.Method == "" {
+			continue // an answer to a change handed over
+		}
+		go p.handle(m)
+	}
+}
+
+func (p *player) handle(m wire) {
+	var a struct {
+		State string          `json:"state"`
+		Key   string          `json:"key"`
+		Value json.RawMessage `json:"value"`
+		Body  json.RawMessage `json:"body"`
+	}
+	_ = json.Unmarshal(m.Params, &a)
+	p.mu.Lock()
+	p.asked = append(p.asked, m.Method+" "+a.State+" "+a.Key)
+	p.mu.Unlock()
+	if m.Method == "mesh/publish" {
+		var e struct {
+			Key  string          `json:"key"`
+			Body json.RawMessage `json:"body"`
+		}
+		_ = json.Unmarshal(m.Params, &e)
+		p.answer(m.ID, map[string]any{}, p.nc.Publish("mesh.mod.disk-load.event."+e.Key, e.Body))
+		return
+	}
+	if !strings.HasPrefix(m.Method, "mesh/state.") {
+		p.answer(m.ID, nil, errors.New("nats: no responders available for request"))
+		return
+	}
+	kv, err := p.js.KeyValue(p.ctx, bucket(a.State))
+	if err != nil {
+		p.answer(m.ID, nil, err)
+		return
+	}
+	switch m.Method {
+	case "mesh/state.get":
+		e, err := kv.Get(p.ctx, a.Key)
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			p.answer(m.ID, nil, nil)
+			return
+		}
+		if err != nil {
+			p.answer(m.ID, nil, err)
+			return
+		}
+		p.answer(m.ID, map[string]any{"key": e.Key(), "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil)
+	case "mesh/state.put":
+		rev, err := kv.Put(p.ctx, a.Key, a.Value)
+		p.answer(m.ID, map[string]any{"revision": rev}, err)
+	case "mesh/state.keys":
+		keys, err := kv.Keys(p.ctx)
+		if errors.Is(err, jetstream.ErrNoKeysFound) {
+			keys, err = []string{}, nil
+		}
+		p.answer(m.ID, keys, err)
+	case "mesh/state.watch":
+		p.watch(m.ID, a.State, kv, a.Key)
+	default:
+		p.answer(m.ID, nil, errors.New("not played here"))
+	}
+}
+
+// watch hands over the bus's watch of the pattern: every current value, then the answer to the watch, then each change.
+func (p *player) watch(id json.RawMessage, state string, kv jetstream.KeyValue, pattern string) {
+	var watchID string
+	_ = json.Unmarshal(id, &watchID)
+	if pattern == "" {
+		pattern = ">"
+	}
+	w, err := kv.Watch(p.ctx, strings.ReplaceAll(pattern, "**", ">"))
+	if err != nil {
+		p.answer(id, nil, err)
+		return
+	}
+	current := true
+	for e := range w.Updates() {
+		if e == nil {
+			current = false
+			p.answer(id, map[string]any{}, nil)
+			continue
+		}
+		op := "put"
+		if e.Operation() != jetstream.KeyValuePut {
+			op = "delete"
+		}
+		p.mu.Lock()
+		p.next++
+		n := p.next
+		p.mu.Unlock()
+		p.send(map[string]any{"jsonrpc": "2.0", "id": fmt.Sprintf("rt-%d", n), "method": "mesh/state", "params": map[string]any{
+			"watch": watchID, "change": map[string]any{"state": state, "key": e.Key(), "op": op,
+				"value": json.RawMessage(e.Value()), "revision": e.Revision(), "current": current}}})
+	}
+}
+
+// 0577/AC5 (the group's integration test): a key of the node-engine's state data written after disk-load started is
+// read within a minute, its item's reading is on the bus, and disk-load read no composition input to learn it.
+func TestDiskLoadWholeMeasuresAnItemWithinAMinuteOfItsKey(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	bin := filepath.Join(t.TempDir(), "disk-load")
+	build := exec.Command("go", "build", "-o", bin, ".")
+	build.Stderr = os.Stderr
+	if err := build.Run(); err != nil {
+		t.Fatalf("disk-load does not build: %v", err)
+	}
+	nc, err := nats.Connect(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, b := range []string{delivery.EngineDataBucket, health.ReadingsBucket, "disk-load_" + StateRounds} {
+		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	cmd := exec.CommandContext(ctx, bin)
+	cmd.Env = append(os.Environ(), "MESH_NODE=ace", "MESH_MODULE=disk-load")
+	stdin, _ := cmd.StdinPipe()
+	stdout, _ := cmd.StdoutPipe()
+	cmd.Stderr = os.Stderr
+	if err := cmd.Start(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = stdin.Close(); cancel(); _ = cmd.Wait() })
+	p := &player{ctx: ctx, nc: nc, js: js, in: stdin}
+	go p.serve(stdout)
+
+	// Once disk-load watches its machine's keys, the engine writes one.
+	watching := func() bool {
+		p.mu.Lock()
+		defer p.mu.Unlock()
+		for _, a := range p.asked {
+			if a == "mesh/state.watch "+delivery.EngineDataRead+" ace.*" {
+				return true
+			}
+		}
+		return false
+	}
+	for end := time.Now().Add(20 * time.Second); !watching(); time.Sleep(50 * time.Millisecond) {
+		if time.Now().After(end) {
+			t.Fatalf("disk-load did not watch its machine's keys of %s: asked %v", delivery.EngineDataBucket, p.asked)
+		}
+	}
+	data, _ := js.KeyValue(ctx, delivery.EngineDataBucket)
+	dir := t.TempDir()
+	item := catalogue.PlacedDataItem{Module: "pg", ID: "data", Class: "valuable", Path: dir, Measure: catalogue.MeasureShallow}
+	raw, err := delivery.EncodeEngineData(delivery.EngineData{Node: "ace", Module: "pg", Digest: "sha256:pg", Items: []catalogue.PlacedDataItem{item}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	key, _ := delivery.EngineDataKey("ace", "pg")
+	written := time.Now()
+	if _, err := data.Put(ctx, key, raw); err != nil {
+		t.Fatal(err)
+	}
+	readings, _ := js.KeyValue(ctx, health.ReadingsBucket)
+	for {
+		e, err := readings.Get(ctx, health.ReadingKey("ace", "pg", "data"))
+		if err == nil {
+			r, err := health.DecodeReading(e.Value())
+			if err != nil || r.Path != dir || r.MeasuredAt == nil {
+				t.Fatalf("the reading is %s (%v)", e.Value(), err)
+			}
+			break
+		}
+		if time.Since(written) > time.Minute {
+			t.Fatalf("no reading within a minute of the key's write: %v", err)
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	// It composed nothing: every state it asked is the engine's data or its own.
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	for _, a := range p.asked {
+		f := strings.Fields(a)
+		if len(f) > 1 && strings.HasPrefix(f[0], "mesh/state.") && f[1] != delivery.EngineDataRead && f[1] != StateReadings && f[1] != StateRounds {
+			t.Errorf("disk-load read %s, which is no state of its own nor the engine's data", a)
+		}
+	}
+}

modules/disk-load/go.mod

@@ -2,13 +2,21 @@ module diskload
 
 go 1.26.0
 
-require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031213356-5ae69df8ccde
+require (
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031230538-c28c645ca406
+	github.com/nats-io/nats.go v1.54.0
+)
 
 require (
+	github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
+	github.com/google/go-tpm v0.9.8 // indirect
 	github.com/klauspost/compress v1.20.0 // indirect
-	github.com/nats-io/nats.go v1.54.0 // indirect
+	github.com/minio/highwayhash v1.0.4 // indirect
+	github.com/nats-io/jwt/v2 v2.8.1 // indirect
+	github.com/nats-io/nats-server/v2 v2.11.17 // indirect
 	github.com/nats-io/nkeys v0.4.16 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	golang.org/x/crypto v0.57.0 // indirect
 	golang.org/x/sys v0.48.0 // indirect
+	golang.org/x/time v0.15.0 // indirect
 )

modules/disk-load/go.sum

@@ -1,7 +1,17 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031213356-5ae69df8ccde h1:RyISctWEfdaykq69I3CYKOZ4N1wUx6PpiPbql4kpXkg=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031230538-c28c645ca406 h1:Z4RcH5Ft4OAAG3F538NXIHP9F3pkyk3j0LtP9mqu1qQ=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031213356-5ae69df8ccde/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031230538-c28c645ca406/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
+github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
+github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
 github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
 github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
+github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
+github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
+github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
+github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
+github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
+github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
 github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
 github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
 github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
@@ -10,5 +20,8 @@ github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
 golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
+golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
 golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
+golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
+golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=

modules/disk-load/module.json

@@ -9,6 +9,7 @@
     "disk_devices",
     "cpu_top",
     "disk_readings",
+    "disk_round",
     "disk_measure"
   ],
   "build": {
@@ -27,14 +28,13 @@
     ]
   },
   "state": [
-    "readings"
+    "readings",
+    {
+      "name": "rounds",
+      "per-machine": true
+    }
   ],
   "reads": [
-    "mesh-config.resolved",
+    "mesh-host.data"
-    "mesh-delivery-assigner.placement",
-    "mesh-delivery-builder.builds",
-    "mesh-delivery-indexer.manifests",
-    "mesh-delivery-indexer.modules",
-    "node-engine.have"
   ]
 }