diff --git a/modules/disk-load/cmd/disk-load/data.go b/modules/disk-load/cmd/disk-load/data.go index 7882c8040..c05a986e2 100644 --- a/modules/disk-load/cmd/disk-load/data.go +++ b/modules/disk-load/cmd/disk-load/data.go @@ -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 `.` per module placed on this node that declares data items, as the +// (engineDataBucket), one key `.` 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 diff --git a/modules/disk-load/cmd/disk-load/data_test.go b/modules/disk-load/cmd/disk-load/data_test.go index 11442ad85..ba748f02b 100644 --- a/modules/disk-load/cmd/disk-load/data_test.go +++ b/modules/disk-load/cmd/disk-load/data_test.go @@ -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 +} diff --git a/modules/disk-load/cmd/disk-load/keeper.go b/modules/disk-load/cmd/disk-load/keeper.go index 5f7d4a93d..7f48cfe61 100644 --- a/modules/disk-load/cmd/disk-load/keeper.go +++ b/modules/disk-load/cmd/disk-load/keeper.go @@ -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 diff --git a/modules/disk-load/cmd/disk-load/packages_test.go b/modules/disk-load/cmd/disk-load/packages_test.go index bea845330..dabc59ef9 100644 --- a/modules/disk-load/cmd/disk-load/packages_test.go +++ b/modules/disk-load/cmd/disk-load/packages_test.go @@ -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 diff --git a/modules/disk-load/cmd/disk-load/whole_integration_test.go b/modules/disk-load/cmd/disk-load/whole_integration_test.go index 4f568cc41..fa33221f6 100644 --- a/modules/disk-load/cmd/disk-load/whole_integration_test.go +++ b/modules/disk-load/cmd/disk-load/whole_integration_test.go @@ -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) } } diff --git a/modules/disk-load/go.mod b/modules/disk-load/go.mod index 75ea37253..5fb0e1ffe 100644 --- a/modules/disk-load/go.mod +++ b/modules/disk-load/go.mod @@ -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 )