diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go new file mode 100644 index 0000000..eeaa473 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go @@ -0,0 +1,366 @@ +//go:build integration + +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the +// controller's user and the builder's own principals . (two instances, for the test of one build). +const busConf = ` +listen: 127.0.0.1:-1 +jetstream { store_dir: %q } +accounts { + MESH { + jetstream: enabled + users = [ + { user: "ctl", password: "pw" } + { user: "n1.mesh-delivery-builder", password: "pw" } + { user: "n2.mesh-delivery-builder", password: "pw" } + ] + } +} +` + +const ( + commit = "7a74cee83f88e6585a5511c3cc53bdf0f5303d20" + fp = "src2:3333333333333333333333333333333333333333333333333333333333333333" +) + +// fakeBuildModules serve the tools of mesh-build-go and mesh-build-node as the real ones do (ADR 0469 clause 2): a call on +// mesh.mod..call... answers at once with the step and its key, and the step's end is written to +// the key . of the module's steps bucket, some milliseconds later. +type fakeBuildModules struct { + js jetstream.JetStream + ctx context.Context + n atomic.Int64 + failBuild atomic.Bool + mu sync.Mutex + calls map[string]int // by `.` +} + +func (f *fakeBuildModules) count(tool string) int { + f.mu.Lock() + defer f.mu.Unlock() + return f.calls[tool] +} + +// digestOf is what every node builds an artifact to. +func digestOf(artifact string) string { + sum := sha256.Sum256([]byte(artifact)) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +func (f *fakeBuildModules) serve(t *testing.T, nc *nats.Conn, module string) { + steps, err := f.js.CreateKeyValue(f.ctx, jetstream.KeyValueConfig{Bucket: delivery.StepsBucket(module), History: 1}) + if err != nil { + t.Fatal(err) + } + sub, err := nc.Subscribe("mesh.mod."+module+".call.>", func(m *nats.Msg) { + tokens := strings.Split(m.Subject, ".") // mesh mod call + tool, node := tokens[4], tokens[5] + f.mu.Lock() + f.calls[module+"."+tool]++ + f.mu.Unlock() + var args map[string]any + _ = json.Unmarshal(m.Data, &args) + id := fmt.Sprintf("s%d", f.n.Add(1)) + key, _ := delivery.StepKey(node, id) + put := func(s delivery.Step) { + data, err := delivery.EncodeStep(s) + if err == nil { + _, err = steps.Put(f.ctx, key, data) + } + if err != nil { + t.Errorf("writing step %s: %v", key, err) + } + } + step := delivery.Step{ID: id, Tool: tool, Node: node, State: delivery.StepRunning, Started: time.Now().UTC()} + put(step) + answer, _ := json.Marshal(map[string]any{"result": map[string]any{"step": id, "key": key, "state": "running"}, "error": "", "node": node}) + _ = m.Respond(answer) + go func() { + time.Sleep(30 * time.Millisecond) + step.Ended = time.Now().UTC() + switch { + case strings.HasSuffix(tool, "_fetch"): + step.State, step.Result = delivery.StepPassed, json.RawMessage(`{"workspace":"w`+id+`","commit":"`+commit+`"}`) + case f.failBuild.Load() && node == "n1": + step.State, step.Reason = delivery.StepFailed, fmt.Sprintf("go build of %v: boom", args["artifact"]) + default: + step.State, step.Result = delivery.StepPassed, json.RawMessage(`{"digest":"`+digestOf(fmt.Sprint(args["artifact"]))+`"}`) + } + put(step) + }() + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) +} + +type bed struct { + ctx context.Context + js jetstream.JetStream + modules jetstream.KeyValue + builds jetstream.KeyValue + tools map[string]jetstream.KeyValue + fake *fakeBuildModules + srvURL string + dir string +} + +// startBed starts the bus, the buckets the controller makes from the builder's and the build modules' manifests, the +// build modules' streams and the builder's durable consumers on them, two nodes' toolchains, the indexer's manifests, and +// the fake build modules. It runs no builder. +func startBed(t *testing.T) *bed { + dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "b") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + conf := filepath.Join(dir, "bus.conf") + if err := os.WriteFile(conf, []byte(fmt.Sprintf(busConf, filepath.Join(dir, "js"))), 0o600); err != nil { + t.Fatal(err) + } + opts, err := server.ProcessConfigFile(conf) + if err != nil { + t.Fatal(err) + } + opts.NoLog, opts.NoSigs = true, true + srv, err := server.NewServer(opts) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(srv.Shutdown) + ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(ctl.Close) + js, err := jetstream.New(ctl) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + t.Cleanup(cancel) + b := &bed{ctx: ctx, js: js, tools: map[string]jetstream.KeyValue{}, srvURL: srv.ClientURL(), dir: dir} + for module, stream := range toolchainStreams { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.stream." + module + ".event.toolchain.reported"}}); err != nil { + t.Fatal(err) + } + for _, node := range []string{"n1", "n2"} { + if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: node + "_mesh-delivery-builder", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + } + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket(module), History: 1}) + if err != nil { + t.Fatal(err) + } + b.tools[module] = kv + } + mk := func(name string, history int) jetstream.KeyValue { + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: uint8(history)}) + if err != nil { + t.Fatal(err) + } + return kv + } + b.modules = mk(delivery.ModulesBucket, delivery.ModulesHistory) + mk(delivery.ManifestsBucket, delivery.ManifestsHistory) + b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory) + b.fake = &fakeBuildModules{js: js, ctx: ctx, calls: map[string]int{}} + b.fake.serve(t, ctl, "mesh-build-go") + b.fake.serve(t, ctl, "mesh-build-node") + b.report(t, "mesh-build-go", "n1", "go", "go1.27.1", "1") + b.report(t, "mesh-build-go", "n2", "go", "go1.27.1", "1") + b.report(t, "mesh-build-node", "n1", "typescript", "tsc5", "1") + b.report(t, "mesh-build-node", "n2", "typescript", "tsc5", "1") + return b +} + +// report writes a node's toolchains key of a build module and says so on the module's stream, as the module does. +func (b *bed) report(t *testing.T, module, node, language, version, recipe string) { + t.Helper() + data, err := delivery.EncodeToolchain(delivery.Toolchain{Language: language, Version: version, Recipe: recipe, At: time.Now().UTC()}) + if err != nil { + t.Fatal(err) + } + key, _ := delivery.ToolchainKey(node, language) + if _, err := b.tools[module].Put(b.ctx, key, data); err != nil { + t.Fatal(err) + } + if _, err := b.js.Publish(b.ctx, "mesh.stream."+module+".event.toolchain.reported", []byte(`{}`)); err != nil { + t.Fatal(err) + } +} + +// module writes modules. as the indexer does for a module with a new-form fingerprint, and its manifest by digest. +func (b *bed) module(t *testing.T, name, fixture, path string) string { + t.Helper() + raw, err := os.ReadFile(filepath.Join("testdata", fixture)) + if err != nil { + t.Fatal(err) + } + digest := delivery.ManifestDigest(raw) + manifests, err := b.js.KeyValue(b.ctx, delivery.ManifestsBucket) + if err != nil { + t.Fatal(err) + } + mv, err := delivery.EncodeManifest(delivery.Manifest{Module: name, JSON: string(raw)}) + if err != nil { + t.Fatal(err) + } + if _, err := manifests.Put(b.ctx, digest, mv); err != nil { + t.Fatal(err) + } + data, err := delivery.EncodeModule(delivery.Module{Repository: "novox/mesh-catalog", Commit: commit, FP: "src1:" + strings.Repeat("f", 64), FP2: fp, + Manifest: digest, Path: path}) + if err != nil { + t.Fatal(err) + } + if _, err := b.modules.Put(b.ctx, name, data); err != nil { + t.Fatal(err) + } + return digest +} + +// start runs the program as the builder of a node. +func (b *bed) start(t *testing.T, node string, env map[string]string) { + t.Helper() + broker := filepath.Join(t.TempDir(), "broker") + cred, _ := json.Marshal(credential{URL: b.srvURL, Node: node, Module: "mesh-delivery-builder", User: node + ".mesh-delivery-builder", Password: "pw"}) + if err := os.WriteFile(broker, cred, 0o600); err != nil { + t.Fatal(err) + } + e := map[string]string{"MESH_BROKER_FILE": broker} + for k, v := range env { + e[k] = v + } + runCtx, stop := context.WithCancel(b.ctx) + done := make(chan error, 1) + go func() { done <- run(runCtx, func(k string) string { return e[k] }, io.Discard) }() + t.Cleanup(func() { stop(); <-done }) +} + +// await reads the build key until pred holds of it, within 20 s. +func (b *bed) await(t *testing.T, module string, pred func(delivery.BuilderBuild) bool) delivery.BuilderBuild { + t.Helper() + key, err := delivery.BuilderBuildKey(module, fp) + if err != nil { + t.Fatal(err) + } + var last delivery.BuilderBuild + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + e, err := b.builds.Get(b.ctx, key) + if err != nil { + continue + } + if last, err = delivery.DecodeBuilderBuild(e.Value()); err == nil && pred(last) { + return last + } + } + t.Fatalf("builds.%s = %+v within 20 s", key, last) + return last +} + +func isState(state string) func(delivery.BuilderBuild) bool { + return func(v delivery.BuilderBuild) bool { return v.State == state } +} + +// TestBuilderMakesADoneKeyWithTheDigests: modules.netcheck (a Go and a TypeScript bundle) and its manifest are in the +// indexer's buckets, and two nodes report each build module's toolchain; the program writes builds.netcheck. done, +// the digests of both bundles, both build modules' toolchains, after a fetch and a build on each of the two nodes. +func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7 + b := startBed(t) + digest := b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck") + b.start(t, "n1", nil) + v := b.await(t, "netcheck", isState(delivery.BuildDone)) + if v.By != "n1.mesh-delivery-builder" || v.Manifest != digest || len(v.Artifacts) != 2 { + t.Fatalf("build = %+v", v) + } + for _, a := range v.Artifacts { + if a.Digest != digestOf(a.Name) || a.Kind != "bundle" || a.Reference != "" { + t.Errorf("artifact = %+v, want the digest every node built and no reference", a) + } + } + if v.Toolchains["go"].Version != "go1.27.1" || v.Toolchains["typescript"].Version != "tsc5" { + t.Errorf("toolchains = %+v", v.Toolchains) + } + for tool, want := range map[string]int{"mesh-build-go.build_go_fetch": 2, "mesh-build-go.build_go_build": 2, "mesh-build-node.build_node_fetch": 2, "mesh-build-node.build_node_build": 2} { + if got := b.fake.count(tool); got != want { + t.Errorf("%s called %d times, want %d (once on each of two nodes)", tool, got, want) + } + } +} + +// TestASecondBuilderInstanceDoesNotBuildTheSameSourceAgain: two builders on two nodes start together on one module key; the +// fingerprint is built once. +func TestASecondBuilderInstanceDoesNotBuildTheSameSourceAgain(t *testing.T) { // 1607/AC7 + b := startBed(t) + b.module(t, "ca-trust", "catalogue-a67f0196-ca-trust.module.json", "modules/ca-trust") + b.start(t, "n1", nil) + b.start(t, "n2", nil) + v := b.await(t, "ca-trust", isState(delivery.BuildDone)) + if len(v.Artifacts) != 1 || v.Artifacts[0].Kind != "archive" || v.Artifacts[0].Digest != digestOf("files") { + t.Fatalf("build = %+v, want the archive's digest", v) + } + if n := b.fake.count("mesh-build-go.build_go_build"); n != 2 { // one instance built it, on two nodes + t.Errorf("build_go_build was called %d times, want 2", n) + } +} + +// TestAFailedKeyIsBuiltAgainAsANewRevisionWhenAToolchainChanges: a build step fails on one node, so the key is failed +// naming it; a toolchain report of a node then wakes the builder, which builds the key again as a new revision of the same +// key, now done. +func TestAFailedKeyIsBuiltAgainAsANewRevisionWhenAToolchainChanges(t *testing.T) { // 1607/AC6, AC7 + b := startBed(t) + b.fake.failBuild.Store(true) + b.module(t, "ca-trust", "catalogue-a67f0196-ca-trust.module.json", "modules/ca-trust") + b.start(t, "n1", nil) + v := b.await(t, "ca-trust", isState(delivery.BuildFailed)) + if !strings.Contains(v.Error, "n1") || !strings.Contains(v.Error, "boom") { + t.Errorf("build = %+v, want failed naming the node and the step's reason", v) + } + key, _ := delivery.BuilderBuildKey("ca-trust", fp) + failed, err := b.builds.Get(b.ctx, key) + if err != nil { + t.Fatal(err) + } + b.fake.failBuild.Store(false) + // The build module's recipe moved on both nodes: the toolchain's entries changed, twice (between the two the nodes disagree). + b.report(t, "mesh-build-go", "n1", "go", "go1.27.1", "2") + b.report(t, "mesh-build-go", "n2", "go", "go1.27.1", "2") + b.await(t, "ca-trust", isState(delivery.BuildDone)) + done, err := b.builds.Get(b.ctx, key) + if err != nil { + t.Fatal(err) + } + if done.Revision() <= failed.Revision() { + t.Errorf("revision %d, was %d: the key was not written as a new revision", done.Revision(), failed.Revision()) + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/testdata/catalogue-a67f0196-ca-trust.module.json b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/testdata/catalogue-a67f0196-ca-trust.module.json new file mode 100644 index 0000000..b3c9dc9 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/testdata/catalogue-a67f0196-ca-trust.module.json @@ -0,0 +1,66 @@ +{ + "module": "ca-trust", + "version": "1", + "slug": "catrust", + "capabilities": [ + "service-manager" + ], + "requires": [ + "internal-acme-ca" + ], + "seats": [ + { + "name": "the-mesh-trust-anchor", + "scope": "node" + } + ], + "claims": [ + { + "name": "the-mesh-trust-anchor", + "scope": "node" + } + ], + "resources": [ + { + "id": "state", + "type": "directory", + "mode": "0700", + "place": ".", + "unknown": "program" + }, + { + "id": "anchor", + "type": "file", + "path": "${dir:state}/anchor", + "mode": "0755", + "template": "files/anchor" + }, + { + "id": "unit", + "type": "file", + "path": "/etc/systemd/system/mesh-ca-trust.service", + "mode": "0644", + "template": "files/mesh-ca-trust.service" + }, + { + "id": "trust", + "type": "service", + "unit": "mesh-ca-trust.service", + "state": "running", + "boot": "enabled", + "restart-on": [ + "anchor", + "unit" + ] + } + ], + "build": { + "artifacts": [ + { + "name": "files", + "kind": "archive", + "from": "files" + } + ] + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/testdata/catalogue-a67f0196-netcheck.module.json b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/testdata/catalogue-a67f0196-netcheck.module.json new file mode 100644 index 0000000..ec01003 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/testdata/catalogue-a67f0196-netcheck.module.json @@ -0,0 +1,57 @@ +{ + "module": "netcheck", + "version": "1", + "emits": [ + "tunnel.endpoint-changed", + "tunnel.handshake-stale", + "tunnel.handshake-fresh" + ], + "tools": [ + "netcheck_tunnel", + "netcheck_tcp", + "netcheck_dns", + "netcheck_http", + "netcheck_listening" + ], + "replaces": { + "netcheck_listening": [ + "ss -lunt", + "ss -ltn", + "netstat -lnt" + ], + "netcheck_dns": [ + "dig", + "nslookup" + ], + "netcheck_tunnel": [ + "wg show", + "wg show mesh0 dump" + ] + }, + "build": { + "artifacts": [ + { + "name": "tools-go", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/netcheck", + "binary": "netcheck", + "loads": [ + "netcheck" + ] + }, + { + "name": "tools-typescript", + "kind": "bundle", + "language": "typescript", + "entrypoints": [ + "tools/index.js" + ], + "loads": [ + "tools/index.js" + ] + } + ] + } +} diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index 8e6c35a..b927a9a 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -41,6 +41,7 @@ type Builder struct { mu sync.Mutex known map[string]delivery.Module // by modules key inflight map[string]bool // by build key + again map[string]delivery.Module // by build key: the newest value that arrived while the key was being built running map[string]int // steps in flight, by `/` plans map[string]planned // by manifest digest } @@ -57,7 +58,7 @@ func (b *Builder) init() { limit = 4 } b.sem = make(chan struct{}, limit) - b.known, b.inflight, b.running, b.plans = map[string]delivery.Module{}, map[string]bool{}, map[string]int{}, map[string]planned{} + b.known, b.inflight, b.again, b.running, b.plans = map[string]delivery.Module{}, map[string]bool{}, map[string]delivery.Module{}, map[string]int{}, map[string]planned{} }) } @@ -147,6 +148,7 @@ func (b *Builder) consider(ctx context.Context, module string, m delivery.Module } b.mu.Lock() if b.inflight[key] { + b.again[key] = m // looked at once more when the build ends, so a change is never lost b.mu.Unlock() return } @@ -155,10 +157,21 @@ func (b *Builder) consider(ctx context.Context, module string, m delivery.Module b.wg.Add(1) go func() { defer b.wg.Done() - defer func() { b.mu.Lock(); delete(b.inflight, key); b.mu.Unlock() }() b.sem <- struct{}{} defer func() { <-b.sem }() - b.try(ctx, module, m) + for { + b.try(ctx, module, m) + b.mu.Lock() + next, more := b.again[key] + delete(b.again, key) + if !more || ctx.Err() != nil { + delete(b.inflight, key) + b.mu.Unlock() + return + } + b.mu.Unlock() + m = next + } }() } diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index fb71751..23427b0 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -22,6 +22,9 @@ type buildModules struct { steps *fakeSteps digests map[string]string // `/` -> digest; the default is `digest` fails map[string]string // `/` -> the reason the step fails with + // hold, when set before the builds start, blocks every call until it is closed; entered is sent to as a call arrives. + hold chan struct{} + entered chan struct{} } func newBuildModules() *buildModules { @@ -29,6 +32,13 @@ func newBuildModules() *buildModules { } func (b *buildModules) CallOn(node, key string, body any) (json.RawMessage, error) { + if b.hold != nil { + select { + case b.entered <- struct{}{}: + default: + } + <-b.hold + } b.mu.Lock() defer b.mu.Unlock() b.calls = append(b.calls, call{node, key, body}) @@ -293,3 +303,20 @@ func TestTwoInstancesBuildOneFingerprintOnce(t *testing.T) { // 1607/AC7 t.Errorf("%d builds, want 2: one instance built it", n) } } + +func TestAModulesValueThatArrivesWhileItsKeyIsBeingBuiltIsLookedAtAgainWhenItEnds(t *testing.T) { // 1607/AC6 (no event is dropped, no timer) + r := newRig(t, goBundle) + r.bm.fails["ace/build_go_build"] = "go build of tool: boom" + r.bm.hold, r.bm.entered = make(chan struct{}), make(chan struct{}, 8) + value := moduleValue(t, "modules/m") + r.b.Module(context.Background(), "m", value) + <-r.bm.entered // the first build is under way + r.b.Module(context.Background(), "m", value) + r.b.Reconsider(context.Background()) + delete(r.bm.fails, "ace/build_go_build") // what the event is about: the build can succeed now + close(r.bm.hold) + r.b.Wait() + if v := valueOf(t, r.st, r.key); v.State != delivery.BuildDone { + t.Errorf("build = %+v, want the failed key built again as a new revision after the first build ended", v) + } +}