diff --git a/go.mod b/go.mod index 1a43aa7..e6cd69a 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 11d9d45..c91c159 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 44ca041..280b777 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -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-` 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. diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 4c68ffe..d8eb1f4 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -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 {