novox/mesh-catalog
18
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

disk-load reads the node-engine's data state through the node-engine client and the seat's types (B5e; ADR 0634 §1, §4; issue 1966)

Browse Source 
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-05 15:28:35 +01:00
parent a609ebc8e5
commit 8fabdc4b91
6 changed files with 61 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/disk-load/cmd/disk-load/data.go
+29 -9
View File
@@ -1,7 +1,7 @@
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
// (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).
@@ -12,11 +12,30 @@ import (
"fmt"
"sync"
engineclient "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// The node-engine's state data as the bus names it: the seat's local name under its holder (the one name of the holder
// here, held equal to the manifest's `reads` by a unit row).
const (
engineHolder = "mesh-host"
engineDataBucket = engineHolder + "_" + nodeengine.StateData
engineDataRead = engineHolder + "." + nodeengine.StateData
)
// itemsOf are the seat's data items as this module measures them.
func itemsOf(in []nodeengine.PlacedDataItem) []catalogue.PlacedDataItem {
out := make([]catalogue.PlacedDataItem, 0, len(in))
for _, it := range in {
out = append(out, catalogue.PlacedDataItem{Module: it.Module, ID: it.Item, Class: it.Class, Path: it.Path,
CoveredBy: it.CoveredBy, Protection: it.Protection, Measure: it.Measure})
}
return out
}
// 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
@@ -29,7 +48,7 @@ type Watched struct {
// 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: " +
var ErrNotWatchedYet = errors.New("the node-engine's state " + 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.
@@ -48,21 +67,22 @@ func (w *Watched) Take(ch stdio.StateChange) (string, bool, error) {
delete(w.byModule, module)
return module, true, nil
}
d, err := delivery.DecodeEngineData(ch.Value)
d, err := engineclient.DecodeData(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 {
items := itemsOf(d.Items)
for _, it := range 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)
changed := !sameItems(w.byModule[d.Module], items)
w.byKey[ch.Key] = d.Module
w.byModule[d.Module] = d.Items
w.byModule[d.Module] = items
return d.Module, changed, nil
}
@@ -102,11 +122,11 @@ func sameItems(a, b []catalogue.PlacedDataItem) bool {
// 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)
pattern, err := engineclient.DataNodeKeys(w.Node)
if err != nil {
return err
}
stdio.State(delivery.EngineDataRead).Follow(pattern, func(ch stdio.StateChange) error {
stdio.State(engineDataRead).Follow(pattern, func(ch stdio.StateChange) error {
module, changed, err := w.Take(ch)
if err != nil || !changed || ch.Current {
return err
modules/disk-load/cmd/disk-load/data_test.go
+14 -3
View File
@@ -10,9 +10,10 @@ import (
"time"
"git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings"
engineclient "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// ADR 0577 clauses 2, 3 (and the disk-load check of clause 5): the items measured come from the node-engine's keys of
@@ -20,9 +21,9 @@ import (
// 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 }
key := func(module string) string { k, _ := engineclient.DataKey("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})
raw, err := engineclient.EncodeData(nodeengine.DataByNodeModuleValue{V: 1, Node: "ace", Module: module, Digest: "sha256:" + module, Items: seatItems(items)})
if err != nil {
t.Fatal(err)
}
@@ -188,3 +189,13 @@ func waitFor(t *testing.T, done func() bool) {
}
t.Fatal("not done within 30 s")
}
// seatItems are measured items as the node-engine's state data keeps them.
func seatItems(in []catalogue.PlacedDataItem) []nodeengine.PlacedDataItem {
out := make([]nodeengine.PlacedDataItem, 0, len(in))
for _, it := range in {
out = append(out, nodeengine.PlacedDataItem{Module: it.Module, Item: it.ID, Class: it.Class, Path: it.Path,
CoveredBy: it.CoveredBy, Protection: it.Protection, Measure: it.Measure})
}
return out
}
modules/disk-load/cmd/disk-load/keeper.go
+1 -2
View File
@@ -8,7 +8,6 @@ import (
"fmt"
"git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"os"
"os/exec"
"sort"
@@ -303,7 +302,7 @@ func (k *Keeper) RetireUnkeyed(ctx context.Context) {
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)
k.Say("the node-engine's state %s holds no key of this machine yet: no reading is retired for it", engineDataBucket)
return
}
var declared []catalogue.PlacedDataItem
modules/disk-load/cmd/disk-load/packages_test.go
+6
View File
@@ -3,6 +3,7 @@ package main
import (
"encoding/json"
"os"
"slices"
"testing"
)
@@ -13,6 +14,7 @@ func TestTheReadingsLayoutIsAPublishedPackageOfThisModule(t *testing.T) {
t.Fatal(err)
}
var m struct {
Reads []string `json:"reads"`
Build struct {
Artifacts []struct {
Name, Kind, Series string
@@ -23,6 +25,10 @@ func TestTheReadingsLayoutIsAPublishedPackageOfThisModule(t *testing.T) {
if err := json.Unmarshal(raw, &m); err != nil {
t.Fatal(err)
}
// The node-engine's data state is read under the one name of its holder this module keeps.
if !slices.Contains(m.Reads, engineDataRead) {
t.Errorf("reads %v do not name %s", m.Reads, engineDataRead)
}
for _, a := range m.Build.Artifacts {
if a.Name == "diskreadings" && a.Kind == "package" && a.Series == "0.1" && len(a.Paths) == 1 && a.Paths[0] == "diskreadings" {
return
modules/disk-load/cmd/disk-load/whole_integration_test.go
+9 -8
View File
@@ -27,8 +27,9 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings"
engineclient "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/testbus"
)
@@ -195,7 +196,7 @@ func TestDiskLoadWholeMeasuresAnItemWithinAMinuteOfItsKey(t *testing.T) {
if err != nil {
t.Fatal(err)
}
for _, b := range []string{delivery.EngineDataBucket, diskreadings.ReadingsBucket, "disk-load_" + StateRounds} {
for _, b := range []string{engineDataBucket, diskreadings.ReadingsBucket, "disk-load_" + StateRounds} {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil {
t.Fatal(err)
}
@@ -217,7 +218,7 @@ func TestDiskLoadWholeMeasuresAnItemWithinAMinuteOfItsKey(t *testing.T) {
p.mu.Lock()
defer p.mu.Unlock()
for _, a := range p.asked {
if a == "mesh/state.watch "+delivery.EngineDataRead+" ace.*" {
if a == "mesh/state.watch "+engineDataRead+" ace.*" {
return true
}
}
@@ -225,17 +226,17 @@ func TestDiskLoadWholeMeasuresAnItemWithinAMinuteOfItsKey(t *testing.T) {
}
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)
t.Fatalf("disk-load did not watch its machine's keys of %s: asked %v", engineDataBucket, p.asked)
}
}
data, _ := js.KeyValue(ctx, delivery.EngineDataBucket)
data, _ := js.KeyValue(ctx, 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}})
raw, err := engineclient.EncodeData(nodeengine.DataByNodeModuleValue{V: 1, Node: "ace", Module: "pg", Digest: "sha256:pg", Items: seatItems([]catalogue.PlacedDataItem{item})})
if err != nil {
t.Fatal(err)
}
key, _ := delivery.EngineDataKey("ace", "pg")
key, _ := engineclient.DataKey("ace", "pg")
written := time.Now()
if _, err := data.Put(ctx, key, raw); err != nil {
t.Fatal(err)
@@ -260,7 +261,7 @@ func TestDiskLoadWholeMeasuresAnItemWithinAMinuteOfItsKey(t *testing.T) {
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 {
if len(f) > 1 && strings.HasPrefix(f[0], "mesh/state.") && f[1] != 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
View File
@@ -3,6 +3,8 @@ module diskload
go 1.26.0
require (
git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f
github.com/nats-io/nats.go v1.54.0
)