Node-engine: the have, the node's facts, the record, the health report and the data items are the seat's contract types, written through the node-engine client; the interim bridge is gone (Have group, ADR 0634 §1, §4; issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
@@ -227,7 +227,7 @@ func (s *Shadow) composeCandidate(ctx context.Context, c *Composer, ask nodeengi
|
||||
case laidErr != nil && baseErr != nil:
|
||||
// The node composes none of its own without the candidate either: the candidate changes nothing of that.
|
||||
a.Verdict, a.Modules = nec.Composed, []nodeengine.Module{}
|
||||
a.Own = nodeengine.Module{Module: delivery.HaveNodeWord, State: nec.ModuleHeld, Words: laidErr.Error()}
|
||||
a.Own = nodeengine.Module{Module: nec.NodeWord, State: nec.ModuleHeld, Words: laidErr.Error()}
|
||||
a.Took = c.now().Sub(began).Milliseconds()
|
||||
return a
|
||||
case laidErr != nil:
|
||||
@@ -244,7 +244,7 @@ func (s *Shadow) composeCandidate(ctx context.Context, c *Composer, ask nodeengi
|
||||
one := func(module string) (nodeengine.Module, bool) {
|
||||
m := nodeengine.Module{Module: module}
|
||||
if module == "" {
|
||||
m.Module = delivery.HaveNodeWord
|
||||
m.Module = nec.NodeWord
|
||||
}
|
||||
if why, held := laidHeld[module]; held {
|
||||
m.State, m.Words = nec.ModuleRefused, why
|
||||
@@ -295,7 +295,7 @@ func (s *Shadow) composeCandidate(ctx context.Context, c *Composer, ask nodeengi
|
||||
if own, ok := one(""); ok {
|
||||
a.Own = own
|
||||
} else {
|
||||
a.Own = nodeengine.Module{Module: delivery.HaveNodeWord, State: nec.ModuleHeld,
|
||||
a.Own = nodeengine.Module{Module: nec.NodeWord, State: nec.ModuleHeld,
|
||||
Words: "the node composes no composition of its own"}
|
||||
}
|
||||
refused = refused || a.Own.State == nec.ModuleRefused
|
||||
|
||||
@@ -3,6 +3,7 @@ package selfcompose
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"github.com/novox/mesh-host/internal/buckets"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
@@ -83,7 +84,7 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) {
|
||||
}
|
||||
place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
Modules: []delivery.PlacedModule{{Module: "lemurs"}}})
|
||||
have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
have, _ := nec.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
watching := func() int { b.mu.Lock(); defer b.mu.Unlock(); return len(b.watches) }
|
||||
@@ -91,9 +92,9 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) {
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
b.put(t, delivery.PlacementBucket, "placement.n1", place)
|
||||
b.put(t, delivery.HaveBucket, "n1.node", have)
|
||||
b.put(t, buckets.Have, "n1.node", have)
|
||||
b.put(t, delivery.ModulesBucket, "lemurs", mod)
|
||||
for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} {
|
||||
for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, buckets.Have, delivery.ModulesBucket, ResolvedBucket} {
|
||||
b.send(bucket, agent.Update{Caught: true})
|
||||
}
|
||||
for !s.ready.Load() && time.Now().Before(deadline) {
|
||||
@@ -106,7 +107,7 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) {
|
||||
}
|
||||
if a.Verdict != nec.Composed || len(a.Modules) != 1 || a.Modules[0].Module != "lemurs" ||
|
||||
a.Modules[0].State != nec.ModuleComposed || a.Modules[0].Fp != fpNew || !a.Modules[0].Changed ||
|
||||
a.Own.Module != delivery.HaveNodeWord || a.Took > nec.AnswerBound.Milliseconds() {
|
||||
a.Own.Module != nec.NodeWord || a.Took > nec.AnswerBound.Milliseconds() {
|
||||
t.Errorf("an accepted candidate answered %+v", a)
|
||||
}
|
||||
// Issue 774: the input it sends the sandbox says how it composed the node (the builds), so the sandbox composes it
|
||||
@@ -181,7 +182,7 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) {
|
||||
s := NewShadow(env, nil)
|
||||
s.c.placements["n1"] = delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
Modules: []delivery.PlacedModule{{Module: "lemurs"}}}
|
||||
s.c.records["n1"] = delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}
|
||||
s.c.records["n1"] = nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}
|
||||
s.c.modules["lemurs"] = delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"}
|
||||
s.c.modules["other"] = delivery.Module{FP2: fp, Manifest: digestOther, Repository: "novox/x", Commit: "c"}
|
||||
s.ready.Store(true)
|
||||
|
||||
@@ -2,6 +2,9 @@ package selfcompose
|
||||
|
||||
import (
|
||||
"context"
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"github.com/novox/mesh-host/internal/buckets"
|
||||
engcomp "github.com/novox/mesh-host/internal/compositions"
|
||||
"slices"
|
||||
"strings"
|
||||
@@ -85,18 +88,18 @@ func TestTheNodePublishesItsCompositions(t *testing.T) {
|
||||
}
|
||||
place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
Modules: []delivery.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}}})
|
||||
have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
|
||||
omod, _ := delivery.EncodeModule(delivery.Module{FP2: ofp, Manifest: odigest, Repository: "novox/o", Commit: "c"})
|
||||
want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true})
|
||||
nodeWant, _ := delivery.EncodeNodeWant(delivery.NodeWant{Self: true})
|
||||
b.put(t, delivery.PlacementBucket, "placement.n1", place)
|
||||
b.put(t, delivery.HaveBucket, "n1.node", have)
|
||||
b.put(t, buckets.Have, "n1.node", have)
|
||||
b.put(t, delivery.ModulesBucket, "lemurs", mod)
|
||||
b.put(t, delivery.ModulesBucket, "oldie", omod)
|
||||
b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want)
|
||||
b.put(t, delivery.AssignerWantsBucket, "want.n1.node", nodeWant)
|
||||
for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket, delivery.BuilderBuildsBucket} {
|
||||
for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, buckets.Have, delivery.ModulesBucket, ResolvedBucket, delivery.BuilderBuildsBucket} {
|
||||
b.send(bucket, agent.Update{Caught: true})
|
||||
}
|
||||
for u := next(t, out); !u.Caught; u = next(t, out) {
|
||||
|
||||
@@ -4,13 +4,15 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"github.com/novox/mesh-host/internal/buckets"
|
||||
engcomp "github.com/novox/mesh-host/internal/compositions"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/novox/mesh-host/internal/compose"
|
||||
"github.com/novox/mesh-host/internal/compose/frombus"
|
||||
)
|
||||
@@ -62,7 +64,7 @@ func (c *Composer) publishData(ctx context.Context) {
|
||||
items, unplaced, held, err := dataItemsOf(ctx, itemsReader{reader{c: c}}, c.env.Mesh, scope)
|
||||
if err != nil {
|
||||
c.sayData("the data items of this node's modules cannot be worked out, and what the engine's state " +
|
||||
delivery.EngineDataBucket + " holds stands; " + c.oweData() + ": " + err.Error())
|
||||
buckets.Data + " holds stands; " + c.oweData() + ": " + err.Error())
|
||||
return
|
||||
}
|
||||
// A module held for a refused manifest declares nothing this pass can read: it is named, its other modules' items
|
||||
@@ -83,7 +85,7 @@ func (c *Composer) publishData(ctx context.Context) {
|
||||
c.mu.Unlock()
|
||||
wrote, removed, err := c.writeData(ctx, items, digests, held)
|
||||
if err != nil {
|
||||
c.sayData("the data items of this node's modules are not all kept in " + delivery.EngineDataBucket +
|
||||
c.sayData("the data items of this node's modules are not all kept in " + buckets.Data +
|
||||
"; " + c.oweData() + ": " + err.Error())
|
||||
return
|
||||
}
|
||||
@@ -93,7 +95,7 @@ func (c *Composer) publishData(ctx context.Context) {
|
||||
c.sayData("")
|
||||
if wrote+removed > 0 {
|
||||
c.env.Say(fmt.Sprintf("the data items of this node's modules: %d key(s) of %s written, %d deleted",
|
||||
wrote, delivery.EngineDataBucket, removed))
|
||||
wrote, buckets.Data, removed))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -145,7 +147,7 @@ func (c *Composer) digestsByModule() map[string]string {
|
||||
func (c *Composer) writeData(ctx context.Context, items map[string][]catalogue.PlacedDataItem, digests map[string]string, held map[string]string) (wrote, removed int, err error) {
|
||||
store := c.env.Data
|
||||
if c.dataKept == nil {
|
||||
pattern, err := delivery.EngineDataNodeKeys(c.env.Node)
|
||||
pattern, err := nodeengineclient.DataNodeKeys(c.env.Node)
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
@@ -176,7 +178,7 @@ func (c *Composer) writeData(ctx context.Context, items map[string][]catalogue.P
|
||||
if len(items[module]) == 0 {
|
||||
continue
|
||||
}
|
||||
key, err := delivery.EngineDataKey(c.env.Node, module)
|
||||
key, err := nodeengineclient.DataKey(c.env.Node, module)
|
||||
if err != nil {
|
||||
failed = append(failed, err.Error())
|
||||
continue
|
||||
@@ -185,7 +187,7 @@ func (c *Composer) writeData(ctx context.Context, items map[string][]catalogue.P
|
||||
if !ok {
|
||||
digest = digests[""] // run inside the node's own composition
|
||||
}
|
||||
raw, err := delivery.EncodeEngineData(delivery.EngineData{Node: c.env.Node, Module: module, Digest: digest, Items: items[module]})
|
||||
raw, err := nodeengineclient.EncodeData(nodeengine.DataByNodeModuleValue{Node: c.env.Node, Module: module, Digest: digest, Items: placedItems(items[module])})
|
||||
if err != nil {
|
||||
failed = append(failed, err.Error())
|
||||
continue
|
||||
@@ -232,9 +234,19 @@ func (c *Composer) writeData(ctx context.Context, items map[string][]catalogue.P
|
||||
// heldKey is whether key is the key of a module held for a refused manifest, which stands as it was.
|
||||
func heldKey(node, key string, held map[string]string) bool {
|
||||
for module := range held {
|
||||
if k, err := delivery.EngineDataKey(node, module); err == nil && k == key {
|
||||
if k, err := nodeengineclient.DataKey(node, module); err == nil && k == key {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// placedItems are a module's data items as the state data carries them.
|
||||
func placedItems(in []catalogue.PlacedDataItem) []nodeengine.PlacedDataItem {
|
||||
out := make([]nodeengine.PlacedDataItem, 0, len(in))
|
||||
for _, i := range in {
|
||||
out = append(out, nodeengine.PlacedDataItem{Module: i.Module, Item: i.ID, Class: i.Class, Path: i.Path,
|
||||
CoveredBy: i.CoveredBy, Protection: i.Protection, Measure: i.Measure})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -3,6 +3,9 @@ package selfcompose
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"github.com/novox/mesh-host/internal/buckets"
|
||||
engcomp "github.com/novox/mesh-host/internal/compositions"
|
||||
"slices"
|
||||
"strings"
|
||||
@@ -86,7 +89,7 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) {
|
||||
var said []string
|
||||
env := Env{Node: "n1", Data: store, Say: func(s string) { said = append(said, s) }}
|
||||
// The engine's grant lets it read its own node's keys, as the listing at a run's first pass does, and no other's.
|
||||
if !grantAllows(t, "n1", delivery.EngineDataBucket, "n1.*") || grantAllows(t, "n1", delivery.EngineDataBucket, "n2.*") {
|
||||
if !grantAllows(t, "n1", buckets.Data, "n1.*") || grantAllows(t, "n1", buckets.Data, "n2.*") {
|
||||
t.Error("the node's grant reads other keys of the engine's state data than its own")
|
||||
}
|
||||
c := New(env)
|
||||
@@ -105,15 +108,15 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) {
|
||||
deletes []string
|
||||
asked int
|
||||
// want is each key's items and digest after the step.
|
||||
want map[string]delivery.EngineData
|
||||
want map[string]nodeengine.DataByNodeModuleValue
|
||||
}{
|
||||
{name: "the run's first pass writes each module's items and deletes what left", due: true,
|
||||
items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}, "oldie": {item("oldie", "x", "/srv/oldie")}, "bare": nil},
|
||||
digests: map[string]string{webKey: "sha256:w1", nodeKey: "sha256:n1"},
|
||||
puts: []string{"n1.oldie", "n1.web"}, deletes: []string{"n1.gone"}, asked: 1,
|
||||
want: map[string]delivery.EngineData{
|
||||
"n1.web": {Module: "web", Digest: "sha256:w1", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db")}},
|
||||
"n1.oldie": {Module: "oldie", Digest: "sha256:n1", Items: []catalogue.PlacedDataItem{item("oldie", "x", "/srv/oldie")}}}},
|
||||
want: map[string]nodeengine.DataByNodeModuleValue{
|
||||
"n1.web": {Module: "web", Digest: "sha256:w1", Items: placedItems([]catalogue.PlacedDataItem{item("web", "db", "/srv/web/db")})},
|
||||
"n1.oldie": {Module: "oldie", Digest: "sha256:n1", Items: placedItems([]catalogue.PlacedDataItem{item("oldie", "x", "/srv/oldie")})}}},
|
||||
{name: "a pass that handed nothing new reads nothing", due: false, asked: 1},
|
||||
{name: "a pass that changes no module's items writes nothing", due: true,
|
||||
items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}, "oldie": {item("oldie", "x", "/srv/oldie")}},
|
||||
@@ -126,8 +129,8 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) {
|
||||
{name: "a module whose composition changed its items is written again", due: false,
|
||||
items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db2")}, "oldie": {item("oldie", "x", "/srv/oldie")}},
|
||||
digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, puts: []string{"n1.web"}, asked: 5,
|
||||
want: map[string]delivery.EngineData{
|
||||
"n1.web": {Module: "web", Digest: "sha256:w2", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db2")}}}},
|
||||
want: map[string]nodeengine.DataByNodeModuleValue{
|
||||
"n1.web": {Module: "web", Digest: "sha256:w2", Items: placedItems([]catalogue.PlacedDataItem{item("web", "db", "/srv/web/db2")})}}},
|
||||
{name: "a module that left the node is deleted", due: true,
|
||||
items: map[string][]catalogue.PlacedDataItem{"oldie": {item("oldie", "x", "/srv/oldie")}},
|
||||
digests: map[string]string{nodeKey: "sha256:n1"}, deletes: []string{"n1.web"}, asked: 6},
|
||||
@@ -136,8 +139,8 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) {
|
||||
items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}},
|
||||
held: map[string]string{"oldie": "the manifest of oldie is refused: a bar item is no icon"},
|
||||
digests: map[string]string{webKey: "sha256:w3", nodeKey: "sha256:n1"}, puts: []string{"n1.web"}, asked: 7,
|
||||
want: map[string]delivery.EngineData{
|
||||
"n1.web": {Module: "web", Digest: "sha256:w3", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db")}}}},
|
||||
want: map[string]nodeengine.DataByNodeModuleValue{
|
||||
"n1.web": {Module: "web", Digest: "sha256:w3", Items: placedItems([]catalogue.PlacedDataItem{item("web", "db", "/srv/web/db")})}}},
|
||||
} {
|
||||
t.Run(step.name, func(t *testing.T) {
|
||||
store.mu.Lock()
|
||||
@@ -159,7 +162,7 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) {
|
||||
t.Errorf("owed a pass: %v after %s; want after %s", c.owed.Load(), wait, step.owed)
|
||||
}
|
||||
for key, want := range step.want {
|
||||
got, err := delivery.DecodeEngineData(store.kv[key])
|
||||
got, err := nodeengineclient.DecodeData(store.kv[key])
|
||||
if err != nil || got.Node != "n1" || got.Module != want.Module || got.Digest != want.Digest ||
|
||||
!slices.Equal(got.Items, want.Items) {
|
||||
t.Errorf("%s holds %+v (%v), want %+v", key, got, err, want)
|
||||
|
||||
@@ -10,7 +10,6 @@ import (
|
||||
engcomp "github.com/novox/mesh-host/internal/compositions"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
)
|
||||
|
||||
// **The packages this node's compositions declare, kept by its engine** (novox/hq ADR 0565 §6, §15; issue 1953). The
|
||||
|
||||
@@ -19,6 +19,9 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"github.com/novox/mesh-host/internal/buckets"
|
||||
"github.com/novox/mesh-host/internal/compositions"
|
||||
engcomp "github.com/novox/mesh-host/internal/compositions"
|
||||
"maps"
|
||||
@@ -76,7 +79,7 @@ type Env struct {
|
||||
Unseal func(sealed string) ([]byte, error)
|
||||
// Held is the sealed values a module's have says it runs, kept by the agent on the bus: what runs after a restart,
|
||||
// before the agent applied anything, and after the switch once the kept declaration is old. By slot
|
||||
// (delivery.Have.SealedAt); legacy is the list a have written before slots holds (delivery.Have.Sealed), reused
|
||||
// (nodeengine.HaveByNodeModuleValue.SealedAt); legacy is the list a have written before slots holds (nodeengine.HaveByNodeModuleValue.Sealed), reused
|
||||
// only within its own module.
|
||||
// An error is a have the bus did not answer for, never one absent: the pass is then tried again (BusFailed).
|
||||
Held func(module string) (slots compositions.Seals, legacy []string, err error)
|
||||
@@ -131,7 +134,7 @@ type Composer struct {
|
||||
// placementKeys is the seat's state placement key by key in either form, each node's placement assembled from it
|
||||
// (delivery.PlacementKeys; research 072 row 37): a write of the holder's under way is never composed half.
|
||||
placementKeys *delivery.PlacementKeys
|
||||
records map[string]delivery.NodeRecord
|
||||
records map[string]nodeengine.NodeRecord
|
||||
// inOwn is, by module, whether this node's have of it names no fingerprint of the new form (mesh-wireguard, a module
|
||||
// at a build of the old form): the node runs it inside its own composition, and composes it there again (issue 1126).
|
||||
inOwn map[string]bool
|
||||
@@ -298,7 +301,7 @@ func New(env Env) *Composer {
|
||||
env.Say = func(string) {}
|
||||
}
|
||||
return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{}, placementKeys: delivery.NewPlacementKeys(),
|
||||
records: map[string]delivery.NodeRecord{}, inOwn: map[string]bool{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{},
|
||||
records: map[string]nodeengine.NodeRecord{}, inOwn: map[string]bool{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{},
|
||||
manifests: map[string][]byte{}, archives: map[string][]byte{}, builds: map[string]delivery.BuilderBuild{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{},
|
||||
pkgs: map[string]compositions.Composed{}, asked: make(chan chan Pass, 8)}
|
||||
}
|
||||
@@ -357,8 +360,8 @@ func (c *Composer) inputs() []input {
|
||||
// Every node's record, and this node's own haves (what it runs inside its own composition): two watches, each
|
||||
// within the node's grant (buscompose: readGrants of the have bucket at `*.node`, NodeKeysGrants at `<node>.>`).
|
||||
// A watch of every key is refused by the bus (issue 1126, 2026-10-27 09:21: the fold's `>` stopped every agent).
|
||||
{delivery.HaveBucket, "*." + delivery.HaveNodeWord, nil, (*Composer).takeRecordChanged},
|
||||
{delivery.HaveBucket, node + ".>", nil, (*Composer).takeModuleHave},
|
||||
{buckets.Have, "*." + nodeengineclient.NodeWord, nil, (*Composer).takeRecordChanged},
|
||||
{buckets.Have, node + ".>", nil, (*Composer).takeModuleHave},
|
||||
{delivery.ModulesBucket, ">", (*Composer).takeModule, nil},
|
||||
{ResolvedBucket, node + ".>", (*Composer).takeValue, nil},
|
||||
// The builder's builds: a pass that held a module for a build not yet done is made again when it lands (issue
|
||||
@@ -690,13 +693,13 @@ func (c *Composer) takePlacement(u agent.Update) {
|
||||
}
|
||||
|
||||
func (c *Composer) takeRecord(u agent.Update) {
|
||||
node, _, isNode, err := delivery.ParseHaveKey(u.Key)
|
||||
node, _, isNode, err := nodeengineclient.ParseHaveKey(u.Key)
|
||||
if err != nil || !isNode {
|
||||
return
|
||||
}
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
h, err := delivery.DecodeHaveNode(u.Value)
|
||||
h, err := nodeengineclient.DecodeHaveNode(u.Value)
|
||||
if u.Deleted || err != nil || h.Record == nil {
|
||||
delete(c.records, node)
|
||||
return
|
||||
@@ -715,13 +718,13 @@ func (c *Composer) takeRecordChanged(u agent.Update) bool {
|
||||
// id; mesh-wireguard: none). Read from the haves, which outlast a restart, never from what the agent has applied
|
||||
// this run: a fresh agent composes before it applies anything (review of the fold, round 1).
|
||||
func (c *Composer) takeModuleHave(u agent.Update) bool {
|
||||
node, module, isNode, err := delivery.ParseHaveKey(u.Key)
|
||||
node, module, isNode, err := nodeengineclient.ParseHaveKey(u.Key)
|
||||
if err != nil || isNode || node != c.env.Node {
|
||||
return false
|
||||
}
|
||||
in := false
|
||||
if !u.Deleted {
|
||||
if h, err := delivery.DecodeHave(u.Value); err == nil {
|
||||
if h, err := nodeengineclient.DecodeHave(u.Value); err == nil {
|
||||
// A have that names nothing it runs, failed (the node could not compose a module it never ran: NotComposed), is
|
||||
// no module run inside the node's own composition: taken as one, it was composed from its manifest past its
|
||||
// build (issue 1126, review round 1 of the builds).
|
||||
@@ -729,9 +732,9 @@ func (c *Composer) takeModuleHave(u agent.Update) bool {
|
||||
// mesh-host #189, novox's postgres 2026-10-31): taken as one, it was composed from its manifest and refused
|
||||
// for its secret reader before the start backed it up again. Any other have naming nothing keeps the rule
|
||||
// (waiting for a yes inside the node's own stays in).
|
||||
interrupted := h.Phase == delivery.PhaseHeld && h.Detail == agent.BackupInterrupted
|
||||
runsNothing := (h.Phase == delivery.PhaseFailed || interrupted) && h.FP == "" && h.Composed == ""
|
||||
in = h.Phase != delivery.PhaseRemoved && !runsNothing && !strings.HasPrefix(h.FP, delivery.SourcePrefix2)
|
||||
interrupted := h.Phase == nodeengineclient.PhaseHeld && h.Detail == agent.BackupInterrupted
|
||||
runsNothing := (h.Phase == nodeengineclient.PhaseFailed || interrupted) && h.Fp == "" && h.Composed == ""
|
||||
in = h.Phase != nodeengineclient.PhaseRemoved && !runsNothing && !strings.HasPrefix(h.Fp, delivery.SourcePrefix2)
|
||||
}
|
||||
}
|
||||
c.mu.Lock()
|
||||
@@ -749,7 +752,7 @@ func (c *Composer) takeModuleHave(u agent.Update) bool {
|
||||
|
||||
// inOwnOf is, of the modules p places, those the node runs inside its own composition (c.inOwn), sorted; c.mu held.
|
||||
// Machines is, from the placements this composer holds, each node by its address on the private network, and the hub's
|
||||
// node: what names the machine a failing network part points at (delivery.NetworkPart.Targets, issue 774). Empty
|
||||
// node: what names the machine a failing network part points at (nodeengine.NetworkPart.Targets, issue 774). Empty
|
||||
// before any placement is read.
|
||||
func (c *Composer) Machines() (byAddress map[string]string, hub string) {
|
||||
c.mu.Lock()
|
||||
@@ -1079,7 +1082,7 @@ func (c *Composer) refused(why string, held map[string]string) {
|
||||
c.refusal = said
|
||||
c.held = heldModules(held)
|
||||
if why != "" {
|
||||
c.held = append(c.held, delivery.HaveNodeWord)
|
||||
c.held = append(c.held, nodeengineclient.NodeWord)
|
||||
}
|
||||
c.mu.Unlock()
|
||||
if !same && c.env.Refused != nil {
|
||||
@@ -1374,10 +1377,10 @@ func (r reader) Placements(context.Context) (map[string]delivery.Placement, erro
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (r reader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
|
||||
func (r reader) Records(context.Context) (map[string]nodeengine.NodeRecord, error) {
|
||||
r.c.mu.Lock()
|
||||
defer r.c.mu.Unlock()
|
||||
out := make(map[string]delivery.NodeRecord, len(r.c.records))
|
||||
out := make(map[string]nodeengine.NodeRecord, len(r.c.records))
|
||||
for k, v := range r.c.records {
|
||||
out[k] = v
|
||||
}
|
||||
|
||||
@@ -8,7 +8,9 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"github.com/novox/mesh-host/internal/buckets"
|
||||
"github.com/novox/mesh-host/internal/compositions"
|
||||
"os"
|
||||
"reflect"
|
||||
@@ -189,19 +191,19 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
omod, _ := delivery.EncodeModule(delivery.Module{FP2: delivery.SourcePrefix2 + strings.Repeat("c", 64), Manifest: odigest, Repository: "novox/o", Commit: "c"})
|
||||
place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
Modules: []delivery.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}}})
|
||||
have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
|
||||
want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true})
|
||||
nodeWant, _ := delivery.EncodeNodeWant(delivery.NodeWant{Self: true})
|
||||
b.put(t, delivery.PlacementBucket, "placement.n1", place)
|
||||
b.put(t, delivery.HaveBucket, "n1.node", have)
|
||||
oldHave, _ := delivery.EncodeHave(delivery.Have{Build: "build-1", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
|
||||
b.put(t, delivery.HaveBucket, "n1.oldie", oldHave)
|
||||
b.put(t, buckets.Have, "n1.node", have)
|
||||
oldHave, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Build: "build-1", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
|
||||
b.put(t, buckets.Have, "n1.oldie", oldHave)
|
||||
b.put(t, delivery.ModulesBucket, "lemurs", mod)
|
||||
b.put(t, delivery.ModulesBucket, "oldie", omod)
|
||||
b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want)
|
||||
b.put(t, delivery.AssignerWantsBucket, "want.n1.node", nodeWant)
|
||||
for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket, delivery.BuilderBuildsBucket} {
|
||||
for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, buckets.Have, delivery.ModulesBucket, ResolvedBucket, delivery.BuilderBuildsBucket} {
|
||||
b.send(bucket, agent.Update{Caught: true})
|
||||
}
|
||||
got := map[string]bool{}
|
||||
@@ -696,10 +698,10 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
bkey, _ := delivery.BuilderBuildKey("tool", next2)
|
||||
tb.kv[delivery.BuilderBuildsBucket+"/"+bkey] = building
|
||||
tb.put(t, delivery.PlacementBucket, "placement.n1", tplace)
|
||||
tb.put(t, delivery.HaveBucket, "n1.node", have)
|
||||
tb.put(t, buckets.Have, "n1.node", have)
|
||||
tb.put(t, delivery.ModulesBucket, "tool", tmod)
|
||||
tb.put(t, delivery.AssignerWantsBucket, "want.n1.tool", twant)
|
||||
for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket,
|
||||
for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, buckets.Have, delivery.ModulesBucket,
|
||||
ResolvedBucket, delivery.BuilderBuildsBucket} {
|
||||
tb.send(bucket, agent.Update{Caught: true})
|
||||
}
|
||||
@@ -1049,7 +1051,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
}
|
||||
// What n1 runs: lemurs and its own composition, as its composer composes them from the trunk.
|
||||
running, _, _, err := frombus.Compose(ctx, fixedReader{placements: map[string]delivery.Placement{"n1": trunk},
|
||||
records: map[string]delivery.NodeRecord{"n1": {Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}},
|
||||
records: map[string]nodeengine.NodeRecord{"n1": {Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}},
|
||||
modules: mods, manifests: map[string][]byte{digest: []byte(lemurs), kdigest: []byte(keeper)}}, frombus.Mesh{}, frombus.Scope{Node: "n1"}, map[string]string{"lemurs": fp})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -1085,10 +1087,10 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
sb.missing(delivery.PlacementCandidatesBucket, false)
|
||||
sb.made(6)
|
||||
rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
rec, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
tp, _ := delivery.EncodePlacement(trunk)
|
||||
sb.put(delivery.PlacementBucket, "placement.n1", tp, 1)
|
||||
sb.put(delivery.HaveBucket, "n1.node", rec, 1)
|
||||
sb.put(buckets.Have, "n1.node", rec, 1)
|
||||
for name, m := range mods {
|
||||
raw, _ := delivery.EncodeModule(m)
|
||||
sb.put(delivery.ModulesBucket, name, raw, 1)
|
||||
@@ -1201,7 +1203,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
}
|
||||
}
|
||||
// The node's own key changing (its answers written) composes nothing again; a group's key deleted drops its answer.
|
||||
sb.put(delivery.HaveBucket, "n1.node", rec, 2)
|
||||
sb.put(buckets.Have, "n1.node", rec, 2)
|
||||
select {
|
||||
case a := <-answered:
|
||||
t.Errorf("answers handed again for nothing changed: %+v", a)
|
||||
@@ -1232,13 +1234,13 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
}
|
||||
}
|
||||
lemursHave := func(detail string, rev uint64) {
|
||||
raw, err := delivery.EncodeHave(delivery.Have{FP: fp, Composed: lemursDigest, Phase: delivery.PhaseRunning,
|
||||
Health: delivery.HealthFailed, Detail: detail})
|
||||
raw, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Composed: lemursDigest, Phase: nodeengineclient.PhaseRunning,
|
||||
Health: nodeengineclient.HealthFailed, Detail: detail})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
hk, _ := delivery.HaveKey("n1", "lemurs")
|
||||
sb.put(delivery.HaveBucket, hk, raw, rev)
|
||||
hk, _ := nodeengineclient.HaveKey("n1", "lemurs")
|
||||
sb.put(buckets.Have, hk, raw, rev)
|
||||
}
|
||||
lemursHave("its tool lemurs_health: no link to the bus is open", 10)
|
||||
drain()
|
||||
@@ -1273,9 +1275,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
t.Errorf("a new revision's answer said %q", got)
|
||||
}
|
||||
settled = answering()
|
||||
hk, _ := delivery.HaveKey("n1", "lemurs")
|
||||
moved, _ := delivery.EncodeHave(delivery.Have{FP: fp, Composed: "sha256:" + strings.Repeat("e", 64), Phase: delivery.PhaseRunning})
|
||||
sb.put(delivery.HaveBucket, hk, moved, 13)
|
||||
hk, _ := nodeengineclient.HaveKey("n1", "lemurs")
|
||||
moved, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Composed: "sha256:" + strings.Repeat("e", 64), Phase: nodeengineclient.PhaseRunning})
|
||||
sb.put(buckets.Have, hk, moved, 13)
|
||||
drain()
|
||||
if got := answering(); got != settled+len(answers) {
|
||||
t.Errorf("what runs moved, and %d answer(s) were composed again for %d group(s)", got-settled, len(answers))
|
||||
@@ -1350,7 +1352,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
go func() { _ = nsh.Run(nctx) }()
|
||||
nb.made(6)
|
||||
nb.put(delivery.PlacementBucket, "placement.n1", tp, 1)
|
||||
nb.put(delivery.HaveBucket, "n1.node", rec, 1)
|
||||
nb.put(buckets.Have, "n1.node", rec, 1)
|
||||
for name, m := range mods {
|
||||
raw, _ := delivery.EncodeModule(m)
|
||||
nb.put(delivery.ModulesBucket, name, raw, 1)
|
||||
@@ -1362,7 +1364,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
oldRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "older", Placement: &trunk})
|
||||
oldKey, _ := delivery.PlacementCandidateKey("older", "n1")
|
||||
nb.put(delivery.PlacementCandidatesBucket, oldKey, oldRaw, 2)
|
||||
for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, delivery.BuilderBuildsBucket} {
|
||||
for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, buckets.Have, delivery.ModulesBucket, delivery.BuilderBuildsBucket} {
|
||||
nb.send(bucket, agent.Update{Caught: true})
|
||||
}
|
||||
select {
|
||||
@@ -1508,7 +1510,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
cc.manifests[digest] = []byte(lemurs)
|
||||
cc.manifests[cdigest] = []byte(withCert)
|
||||
cc.modules = map[string]delivery.Module{"lemurs": {Manifest: digest}, "web": {Manifest: cdigest}}
|
||||
cc.records["n1"] = delivery.NodeRecord{Node: "n1", ServingKey: base64.StdEncoding.EncodeToString(make([]byte, 32))}
|
||||
cc.records["n1"] = nodeengine.NodeRecord{Node: "n1", ServingKey: base64.StdEncoding.EncodeToString(make([]byte, 32))}
|
||||
onlyLemurs := delivery.Placement{Node: "n1", Modules: []delivery.PlacedModule{{Module: "lemurs"}}}
|
||||
cc.certifyIfWanted(context.Background(), reader{c: cc}, onlyLemurs)
|
||||
if certAsks != 0 {
|
||||
@@ -1656,7 +1658,7 @@ func (b *patternBus) caught() {
|
||||
// fixedReader is frombus's Reader over values held, for composing what a node runs.
|
||||
type fixedReader struct {
|
||||
placements map[string]delivery.Placement
|
||||
records map[string]delivery.NodeRecord
|
||||
records map[string]nodeengine.NodeRecord
|
||||
modules map[string]delivery.Module
|
||||
manifests map[string][]byte
|
||||
}
|
||||
@@ -1664,7 +1666,7 @@ type fixedReader struct {
|
||||
func (r fixedReader) Placements(context.Context) (map[string]delivery.Placement, error) {
|
||||
return r.placements, nil
|
||||
}
|
||||
func (r fixedReader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
|
||||
func (r fixedReader) Records(context.Context) (map[string]nodeengine.NodeRecord, error) {
|
||||
return r.records, nil
|
||||
}
|
||||
func (r fixedReader) Modules(context.Context) (map[string]delivery.Module, error) {
|
||||
@@ -1700,34 +1702,34 @@ func (r fixedReader) Served(context.Context, string, string, []string) (map[stri
|
||||
func TestAHaveSaysWhetherItsModuleRunsInsideTheNodesOwn(t *testing.T) {
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
have delivery.Have
|
||||
have nodeengine.HaveByNodeModuleValue
|
||||
in bool
|
||||
// heldToday says the shadow counts the module held today (issue 1126: a seed holding it too holds it as today).
|
||||
heldToday bool
|
||||
}{
|
||||
{"failed, naming nothing it runs", delivery.Have{Phase: delivery.PhaseFailed, Health: delivery.HealthFailed,
|
||||
{"failed, naming nothing it runs", nodeengine.HaveByNodeModuleValue{Phase: nodeengineclient.PhaseFailed, Health: nodeengineclient.HealthFailed,
|
||||
Detail: "the node cannot compose it, and keeps running what it ran: no build"}, false, true},
|
||||
{"failed at a build fingerprint", delivery.Have{FP: "build-0123abcd", Composed: "sha256:" + strings.Repeat("c", 64), Phase: delivery.PhaseFailed,
|
||||
Health: delivery.HealthFailed, Detail: "the node cannot compose it, and keeps running what it ran: no build"}, true, true},
|
||||
{"failed at a build fingerprint", nodeengine.HaveByNodeModuleValue{Fp: "build-0123abcd", Composed: "sha256:" + strings.Repeat("c", 64), Phase: nodeengineclient.PhaseFailed,
|
||||
Health: nodeengineclient.HealthFailed, Detail: "the node cannot compose it, and keeps running what it ran: no build"}, true, true},
|
||||
// The live have of novox's de-spiegel (2026-10-28): a build of the old form, named by `build`, and no `fp`.
|
||||
{"running a build of the old form", delivery.Have{Build: "build-1790368817085312877", Phase: delivery.PhaseRunning, Health: delivery.HealthOK}, true, true},
|
||||
{"running a build of the new form", delivery.Have{FP: delivery.SourcePrefix2 + strings.Repeat("a", 64), Composed: "sha256:" + strings.Repeat("c", 64),
|
||||
Phase: delivery.PhaseRunning, Health: delivery.HealthOK}, false, false},
|
||||
{"running with no fingerprint (mesh-wireguard)", delivery.Have{Phase: delivery.PhaseRunning, Health: delivery.HealthOK}, true, false},
|
||||
{"held today", delivery.Have{FP: "build-0123abcd", Composed: "sha256:" + strings.Repeat("c", 64), Phase: delivery.PhaseHeld,
|
||||
Health: delivery.HealthWaits, Detail: "held: a required config value is unset"}, true, true},
|
||||
{"running a build of the old form", nodeengine.HaveByNodeModuleValue{Build: "build-1790368817085312877", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}, true, true},
|
||||
{"running a build of the new form", nodeengine.HaveByNodeModuleValue{Fp: delivery.SourcePrefix2 + strings.Repeat("a", 64), Composed: "sha256:" + strings.Repeat("c", 64),
|
||||
Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}, false, false},
|
||||
{"running with no fingerprint (mesh-wireguard)", nodeengine.HaveByNodeModuleValue{Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}, true, false},
|
||||
{"held today", nodeengine.HaveByNodeModuleValue{Fp: "build-0123abcd", Composed: "sha256:" + strings.Repeat("c", 64), Phase: nodeengineclient.PhaseHeld,
|
||||
Health: nodeengineclient.HealthWaits, Detail: "held: a required config value is unset"}, true, true},
|
||||
// novox's postgres (2026-10-31), its have written before mesh-host #189: held by a backup the node agent's stop
|
||||
// interrupted and naming nothing; counted in, it was composed from its manifest and refused for its secret reader.
|
||||
{"held by an interrupted backup, naming nothing", delivery.Have{Phase: delivery.PhaseHeld, Health: delivery.HealthWaits,
|
||||
{"held by an interrupted backup, naming nothing", nodeengine.HaveByNodeModuleValue{Phase: nodeengineclient.PhaseHeld, Health: nodeengineclient.HealthWaits,
|
||||
Detail: agent.BackupInterrupted}, false, true},
|
||||
} {
|
||||
t.Run(row.name, func(t *testing.T) {
|
||||
c := New(Env{Node: "n1"})
|
||||
key, err := delivery.HaveKey("n1", "web")
|
||||
key, err := nodeengineclient.HaveKey("n1", "web")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value, err := delivery.EncodeHave(row.have)
|
||||
value, err := nodeengineclient.EncodeHave(row.have)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -1783,16 +1785,16 @@ func TestAnUnreadSeatListHoldsWhatIsNewAndHandsItAtTheRead(t *testing.T) {
|
||||
}
|
||||
place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
Modules: []delivery.PlacedModule{{Module: "lemurs"}}})
|
||||
have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
|
||||
want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true})
|
||||
nodeWant, _ := delivery.EncodeNodeWant(delivery.NodeWant{Self: true})
|
||||
b.put(t, delivery.PlacementBucket, "placement.n1", place)
|
||||
b.put(t, delivery.HaveBucket, "n1.node", have)
|
||||
b.put(t, buckets.Have, "n1.node", have)
|
||||
b.put(t, delivery.ModulesBucket, "lemurs", mod)
|
||||
b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want)
|
||||
b.put(t, delivery.AssignerWantsBucket, "want.n1.node", nodeWant)
|
||||
for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket, delivery.BuilderBuildsBucket} {
|
||||
for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, buckets.Have, delivery.ModulesBucket, ResolvedBucket, delivery.BuilderBuildsBucket} {
|
||||
b.send(bucket, agent.Update{Caught: true})
|
||||
}
|
||||
handed := map[string]bool{}
|
||||
|
||||
@@ -5,7 +5,9 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"github.com/novox/mesh-host/internal/buckets"
|
||||
engcomp "github.com/novox/mesh-host/internal/compositions"
|
||||
"sort"
|
||||
"strings"
|
||||
@@ -160,11 +162,11 @@ func (s *Shadow) Run(ctx context.Context) error {
|
||||
ins := []input{
|
||||
{delivery.PlacementCandidatesBucket, ">", nil, nil},
|
||||
{delivery.PlacementBucket, ">", (*Composer).takePlacement, nil},
|
||||
{delivery.HaveBucket, "*." + delivery.HaveNodeWord, nil, nil},
|
||||
{buckets.Have, "*." + nodeengineclient.NodeWord, nil, nil},
|
||||
{delivery.ModulesBucket, ">", (*Composer).takeModule, nil},
|
||||
{ResolvedBucket, node + ".>", (*Composer).takeValue, nil},
|
||||
// What this node runs moved: an apply ends in its module's have.
|
||||
{delivery.HaveBucket, node + ".>", nil, nil},
|
||||
{buckets.Have, node + ".>", nil, nil},
|
||||
}
|
||||
type tagged struct {
|
||||
in int
|
||||
@@ -315,7 +317,7 @@ func (s *Shadow) takeCandidate(u agent.Update) bool {
|
||||
// takeRecord takes a node's record, and says whether it changed: the node's own key changes with each answer written,
|
||||
// which moves nothing composed.
|
||||
func (s *Shadow) takeRecord(u agent.Update) bool {
|
||||
node, _, isNode, err := delivery.ParseHaveKey(u.Key)
|
||||
node, _, isNode, err := nodeengineclient.ParseHaveKey(u.Key)
|
||||
if err != nil || !isNode {
|
||||
return false
|
||||
}
|
||||
@@ -327,7 +329,7 @@ func (s *Shadow) takeRecord(u agent.Update) bool {
|
||||
c.mu.Lock()
|
||||
after, has := c.records[node]
|
||||
c.mu.Unlock()
|
||||
return had != has || (has && !delivery.SameNodeRecord(before, after))
|
||||
return had != has || (has && !nodeengineclient.SameNodeRecord(before, after))
|
||||
}
|
||||
|
||||
// How long the shadow waits before it says it is not ready, and how long a proposal's answer runs before it is said slow.
|
||||
@@ -600,14 +602,14 @@ func (s *Shadow) heldToday() map[string]string {
|
||||
// takeRan keeps the digest one of this node's module haves says runs, and says whether it changed: a have whose health
|
||||
// or words alone moved runs the same.
|
||||
func (s *Shadow) takeRan(u agent.Update) bool {
|
||||
node, module, isNode, err := delivery.ParseHaveKey(u.Key)
|
||||
node, module, isNode, err := nodeengineclient.ParseHaveKey(u.Key)
|
||||
if err != nil || isNode || node != s.c.env.Node {
|
||||
return false
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
before, had := s.ran[module]
|
||||
h, err := delivery.DecodeHave(u.Value)
|
||||
h, err := nodeengineclient.DecodeHave(u.Value)
|
||||
if s.heldNow == nil {
|
||||
s.heldNow = map[string]string{}
|
||||
}
|
||||
@@ -615,18 +617,18 @@ func (s *Shadow) takeRan(u agent.Update) bool {
|
||||
s.heldNow[module] = ""
|
||||
if !u.Deleted && err == nil {
|
||||
switch {
|
||||
case h.Phase == delivery.PhaseHeld:
|
||||
case h.Phase == nodeengineclient.PhaseHeld:
|
||||
s.heldNow[module] = "held"
|
||||
case h.Phase == delivery.PhaseFailed:
|
||||
case h.Phase == nodeengineclient.PhaseFailed:
|
||||
s.heldNow[module] = "failed"
|
||||
case h.Phase != delivery.PhaseRemoved && h.FP == "" && h.Build != "":
|
||||
case h.Phase != nodeengineclient.PhaseRemoved && h.Fp == "" && h.Build != "":
|
||||
// A build of the old form: its have names a `build` and no `fp`, which no registered build of the trunk
|
||||
// names, so the controller keeps it as it runs (issue 1126: novox's de-spiegel at build-…); a seed keeps it so.
|
||||
s.heldNow[module] = "runs a build of the old form"
|
||||
}
|
||||
}
|
||||
holdMoved := wasHeld != s.heldNow[module] // a hold's start or end changes the answer too
|
||||
if u.Deleted || err != nil || h.Composed == "" || h.Phase == delivery.PhaseRemoved {
|
||||
if u.Deleted || err != nil || h.Composed == "" || h.Phase == nodeengineclient.PhaseRemoved {
|
||||
delete(s.ran, module)
|
||||
return had || holdMoved
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user