novox/mesh-host
18
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge pull request 'The node-engine composes again when a build it waits on is done (issue 774)' (#181) from fix/build-not-done-holds-one-module into main

Browse Source 
Reviewed-on: #181
This commit was merged in pull request #181.
This commit is contained in:
jschoubben
2026-10-30 18:53:08 +00:00
parent 2ae7c53376 33131dcbfe
commit b133a8949d
4 changed files with 126 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module github.com/novox/mesh-host
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030182415-754edfae6924
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030184532-9685990a15ed
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
golang.org/x/crypto v0.57.0
go.sum
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030182415-754edfae6924 h1:aIiKj066gYFEDokUOq0i2n4vsygu5g0XMKJRRaMI4fM=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030182415-754edfae6924/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030184532-9685990a15ed h1:0h1S5YXrMdkjhB7gWzugqdLxb/HgDHWDmaVCNbEDAzc=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030184532-9685990a15ed/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=
internal/selfcompose/selfcompose.go
+24
View File
@@ -180,9 +180,33 @@ func (c *Composer) inputs() []input {
{delivery.HaveBucket, 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
// 774, 2026-10-30 19:30: node-tools' build landed and ace, g14 and shanks stayed refused until another input moved).
{delivery.BuilderBuildsBucket, ">", nil, (*Composer).takeBuildDone},
}
}
// takeBuildDone says whether a build that landed is one this node composes from: done, of a module it wants at that
// build or whose trunk's build it is. Every other build, and a build not done, moves nothing.
func (c *Composer) takeBuildDone(u agent.Update) bool {
if u.Deleted {
return false
}
module, fp, err := delivery.ParseBuilderBuildKey(u.Key)
if err != nil {
return false
}
c.mu.Lock()
w, wanted := c.wants[module]
m := c.modules[module]
c.mu.Unlock()
if !wanted || (w.FP != fp && m.FP2 != fp) {
return false
}
b, err := delivery.DecodeBuilderBuild(u.Value)
return err == nil && b.State == delivery.BuildDone
}
// Watch starts the composer and answers the compositions it makes: each as a key of `composed-<node>` would be, then a
// Caught update once every input was read and the node composed once, then every change. It ends when ctx ends or an
// input's watch ends.
internal/selfcompose/selfcompose_test.go
+99 -2
View File
@@ -170,6 +170,8 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
}
return secretChanged, func() {}, nil
}
var composed atomic.Int64 // passes made: each reads what runs once
env.Running = func() []delivery.Composed { composed.Add(1); return nil }
c := New(env)
out, err := c.Watch(ctx, "n1.>")
if err != nil {
@@ -197,7 +199,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
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} {
for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket, delivery.BuilderBuildsBucket} {
b.send(bucket, agent.Update{Caught: true})
}
got := map[string]bool{}
@@ -428,6 +430,101 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":1,"source":"node"}`))
quiet(t, out)
// Issue 774, 2026-10-30 19:30: node-tools' want moved to a build not yet done, and the node stayed refused after it
// landed until another input moved. A done build of a module the node wants, at its want's or the trunk's build,
// makes a pass; a build not done, of another build or of a module not wanted, makes none.
passes := func() int64 { return composed.Load() }
for _, row := range []struct {
name, module, fp, state string
pass bool
}{
{"lemurs's wanted build, not done", "lemurs", fp, "building", false},
{"lemurs's wanted build lands", "lemurs", fp, delivery.BuildDone, true},
{"another build of lemurs", "lemurs", delivery.SourcePrefix2 + strings.Repeat("e", 64), delivery.BuildDone, false},
{"a module the node does not want", "oldie", delivery.SourcePrefix2 + strings.Repeat("c", 64), delivery.BuildDone, false},
} {
before := passes()
key, _ := delivery.BuilderBuildKey(row.module, row.fp)
build := delivery.BuilderBuild{State: row.state}
if row.state == delivery.BuildDone {
build.Manifest = digest
}
raw, err := delivery.EncodeBuilderBuild(build)
if err != nil {
t.Fatal(err)
}
b.put(t, delivery.BuilderBuildsBucket, key, raw)
quiet(t, out)
if got := passes() > before; got != row.pass {
t.Errorf("%s: a pass made %v, want %v", row.name, got, row.pass)
}
}
// And today's sequence whole: tool runs at its old build; its want and the trunk move to a build not yet done, and
// the node holds tool, keeping what runs, naming why; the build lands, and with no other change tool is handed at it.
{
tb := newFakeBus()
tool := `{"module":"tool","version":"1","capabilities":["systemd"],"build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]},` +
`"resources":[{"id":"service","type":"service","unit":"tool.service","state":"running","boot":"enabled"}]}`
tdigest := delivery.ManifestDigest([]byte(tool))
tman, _ := delivery.EncodeManifest(delivery.Manifest{Module: "tool", JSON: tool})
tkey, _ := delivery.ManifestsKey(tdigest)
tb.kv[delivery.ManifestsBucket+"/"+tkey] = tman
old, next2 := delivery.SourcePrefix2+strings.Repeat("1", 64), delivery.SourcePrefix2+strings.Repeat("2", 64)
tenv := tb.env()
var held []string
var hmu sync.Mutex
tenv.Refused = func(why string, h map[string]string) {
hmu.Lock()
defer hmu.Unlock()
held = append(held, why+fmt.Sprint(h))
}
tenv.Running = func() []delivery.Composed {
return []delivery.Composed{{Node: "n1", Module: "tool", FP: old, Resources: []json.RawMessage{
json.RawMessage(`{"id":"service","module":"tool","type":"service","unit":"tool.service","state":"running","boot":"enabled"}`)}}}
}
tc := New(tenv)
tout, err := tc.Watch(ctx, "n1.>")
if err != nil {
t.Fatal(err)
}
tplace, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
Modules: []delivery.PlacedModule{{Module: "tool"}}})
tmod, _ := delivery.EncodeModule(delivery.Module{FP2: next2, Manifest: tdigest, Repository: "novox/t", Commit: "c"})
twant, _ := delivery.EncodeWant(delivery.Want{FP: next2, Self: true})
building, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: "building"})
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, 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,
ResolvedBucket, delivery.BuilderBuildsBucket} {
tb.send(bucket, agent.Update{Caught: true})
}
toolKey, _ := delivery.ComposedKey("n1", "tool", next2)
for u := next(t, tout); !u.Caught; u = next(t, tout) {
if u.Key == toolKey {
t.Errorf("tool is handed at a build not yet done")
}
}
hmu.Lock()
if len(held) == 0 || !strings.Contains(held[len(held)-1], "no done build yet") {
t.Errorf("the node was told %q, want tool held for its build not yet done", held)
}
hmu.Unlock()
done, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: tdigest,
Artifacts: []delivery.BuiltArtifact{{Name: "files", Kind: "archive", Reference: "artifact-store://tool/files/blobs/sha256:" + strings.Repeat("9", 64),
Digest: "sha256:" + strings.Repeat("9", 64)}}})
tb.mu.Lock()
tb.kv[delivery.BuilderBuildsBucket+"/"+bkey] = done
tb.mu.Unlock()
tb.put(t, delivery.BuilderBuildsBucket, bkey, done)
if u := next(t, tout); u.Key != toolKey || u.Deleted {
t.Errorf("the build landed and the node handed %s, want tool at its new build", u.Key)
}
}
// A want removed takes its composition back.
removed, _ := delivery.EncodeWant(delivery.Want{Removed: true})
b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", removed)
@@ -1081,7 +1178,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, ResolvedBucket} {
for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket, delivery.BuilderBuildsBucket} {
nb.send(bucket, agent.Update{Caught: true})
}
select {